Performance Evaluation of Hudi 0.14 and Spark 3.4.1: Record Level Index vs. Global Bloom & Global Simple Indexes¶
Setup Elastic Stack Locally¶
https://github.com/deviantony/docker-elk
git clone https://github.com/deviantony/docker-elk.git
docker-compose up setup
docker-compose up
http://localhost:5601/app/home#/
username elastic
password changemeBackend¶
import json
import requests
from flask import Flask, request
from datetime import datetime
app = Flask(__name__)
# Elasticsearch server configuration
elasticsearch_url = "http://localhost:9200"
index_name = "hudi" # Use your desired index name
username = "elastic"
password = "changeme"
@app.route('/', methods=["POST"])
def index():
json_data = request.json
print(json_data)
print(type(json_data))
# Add an "ingestionTime" field with the current datetime
json_data["ingestionTime"] = datetime.now().isoformat()
# Ingest the JSON data into Elasticsearch using a PUT request
put_url = f"{elasticsearch_url}/{index_name}/_doc/{json_data['commitTime']}"
headers = {'Content-Type': 'application/json'}
# Provide Elasticsearch credentials
auth = (username, password)
# Send the PUT request to insert the JSON data
response = requests.put(put_url, data=json.dumps(json_data), headers=headers, auth=auth)
if response.status_code == 201:
return 'Data Ingested'
else:
return f'Failed to insert data. Status code: {response.status_code}'
if __name__ == "__main__":
app.run(debug=True)Step 1: Define imports¶
In [1]:
try:
import os
import sys
import uuid
import pyspark
from pyspark.sql import SparkSession
from pyspark import SparkConf, SparkContext
from faker import Faker
print("Imports loaded ")
except Exception as e:
print("error",e)
Imports loaded
Step 2: Create Spark Session¶
In [2]:
HUDI_VERSION = '0.14.0'
SPARK_VERSION = '3.4'
SUBMIT_ARGS = f"--packages org.apache.hudi:hudi-spark{SPARK_VERSION}-bundle_2.12:{HUDI_VERSION} pyspark-shell"
os.environ["PYSPARK_SUBMIT_ARGS"] = SUBMIT_ARGS
os.environ['PYSPARK_PYTHON'] = sys.executable
spark = SparkSession.builder \
.config('spark.serializer', 'org.apache.spark.serializer.KryoSerializer') \
.config('spark.sql.extensions', 'org.apache.spark.sql.hudi.HoodieSparkSessionExtension') \
.config('className', 'org.apache.hudi') \
.config('spark.sql.hive.convertMetastoreParquet', 'false') \
.getOrCreate()
Warning: Ignoring non-Spark config property: className
:: loading settings :: url = jar:file:/Users/soumilnitinshah/anaconda3/lib/python3.11/site-packages/pyspark/jars/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml
Ivy Default Cache set to: /Users/soumilnitinshah/.ivy2/cache The jars for the packages stored in: /Users/soumilnitinshah/.ivy2/jars org.apache.hudi#hudi-spark3.4-bundle_2.12 added as a dependency :: resolving dependencies :: org.apache.spark#spark-submit-parent-38b74d3e-4071-4678-908e-f0f675975cd0;1.0 confs: [default] found org.apache.hudi#hudi-spark3.4-bundle_2.12;0.14.0 in central :: resolution report :: resolve 216ms :: artifacts dl 6ms :: modules in use: org.apache.hudi#hudi-spark3.4-bundle_2.12;0.14.0 from central in [default] --------------------------------------------------------------------- | | modules || artifacts | | conf | number| search|dwnlded|evicted|| number|dwnlded| --------------------------------------------------------------------- | default | 1 | 0 | 0 | 0 || 1 | 0 | --------------------------------------------------------------------- :: retrieving :: org.apache.spark#spark-submit-parent-38b74d3e-4071-4678-908e-f0f675975cd0 confs: [default] 0 artifacts copied, 1 already retrieved (0kB/10ms) 23/10/29 14:00:42 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Data Generator Class¶
In [3]:
import faker
import uuid
import random
global faker
import datetime
import time
faker = Faker()
class DataGenerator(object):
@staticmethod
def get_data(samples):
return [
(
uuid.uuid4().__str__(),
faker.name(),
faker.random_element(elements=('IT', 'HR', 'Sales', 'Marketing')),
faker.state(),
str(faker.random_int(min=10000, max=150000)),
str(faker.random_int(min=18, max=60)),
str(faker.random_int(min=0, max=100000)),
str(faker.unix_time()),
faker.email(),
faker.credit_card_number(card_type='amex'),
random.choice(["2021","2022","2023"]),
faker.month()
) for x in range(samples)
]
Creating Spark DF for Testing with 0.1M samples¶
In [4]:
data = DataGenerator.get_data(samples=100000)
columns = ["emp_id", "employee_name", "department", "state", "salary", "age", "bonus", "ts", "email", "credit_card","year", "month"]
spark_df = spark.createDataFrame(data=data, schema=columns)
spark_df.count()
23/10/29 14:01:07 WARN TaskSetManager: Stage 0 contains a task of very large size (1197 KiB). The maximum recommended task size is 1000 KiB.
Out[4]:
100000
Helper Methods¶
In [5]:
def write_to_hudi(spark_df, table_name, method, index_type='BLOOM'):
path = f"file:///Users/soumilnitinshah/Downloads/hudidb/{table_name}"
hudi_options = {
'hoodie.table.name': table_name,
'hoodie.datasource.write.table.type': 'COPY_ON_WRITE',
'hoodie.datasource.write.table.name': table_name,
'hoodie.datasource.write.operation': method,
'hoodie.datasource.write.recordkey.field': 'emp_id',
'hoodie.datasource.write.precombine.field': 'ts',
'hoodie.datasource.write.partitionpath.field': 'state',
'hoodie.write.commit.callback.http.url': 'http://127.0.0.1:5000',
'hoodie.write.commit.callback.on': 'true',
'hoodie.write.commit.callback.http.timeout.seconds': '300'
}
if index_type == 'RECORD_INDEX':
hudi_options['hoodie.index.type'] = index_type
hudi_options['hoodie.metadata.record.index.enable'] = 'true'
else:
hudi_options['hoodie.index.type'] = index_type
spark_df.write.format("hudi").options(**hudi_options).mode("append").save(path)
def process_and_write_to_hudi(spark_df, table_name, index_type='BLOOM'):
write_to_hudi(spark_df, table_name, method="upsert", index_type=index_type)
def process_and_delete_from_hudi(table_name, condition, index_type='BLOOM'):
path = f"file:///Users/soumilnitinshah/Downloads/hudidb/{table_name}"
spark.read.format("hudi").load(path).createOrReplaceTempView("hudi_snapshot")
delete_df = spark.sql(f"SELECT * FROM hudi_snapshot WHERE {condition}")
write_to_hudi(delete_df, table_name, method="delete", index_type=index_type)
def process_and_update_hudi(table_name, condition, updates, index_type='BLOOM'):
path = f"file:///Users/soumilnitinshah/Downloads/hudidb/{table_name}"
spark.read.format("hudi").load(path).createOrReplaceTempView("hudi_snapshot")
update_df = spark.sql(f"SELECT * FROM hudi_snapshot WHERE {condition}")
# Apply updates to the DataFrame update_df as needed
write_to_hudi(update_df, table_name, method="upsert", index_type=index_type)
Perform Test¶
In [6]:
# Define a list of table configurations
table_configs = [
{"table_name": "global_bloom_index", "index_type": "GLOBAL_BLOOM"},
{"table_name": "global_simple_index", "index_type": "GLOBAL_SIMPLE"},
{"table_name": "record_level_index", "index_type": "RECORD_INDEX"},
]
for config in table_configs:
process_and_write_to_hudi(spark_df, config["table_name"], config["index_type"])
delete_configs = [
{
"table_name": "global_bloom_index",
"condition": "salary >= 50000",
},
{
"table_name": "global_simple_index",
"condition": "salary >= 50000",
},
{
"table_name": "record_level_index",
"condition": "salary >= 50000",
}
]
update_configs = [
{
"table_name": "global_bloom_index",
"condition": "salary >= 5000",
"updates": {"salary": "salary + 100"},
},
{
"table_name": "global_simple_index",
"condition": "salary >= 5000",
"updates": {"salary": "salary + 100"},
},
{
"table_name": "record_level_index",
"condition": "salary >= 5000",
"updates": {"salary": "salary + 100"},
}
]
# Loop through deletion configurations and process/delete from Hudi
for config in delete_configs:
process_and_delete_from_hudi(config["table_name"], config["condition"])
# Loop through update configurations and process/update Hudi
for config in update_configs:
process_and_update_hudi(config["table_name"], config["condition"], config["updates"])
23/10/29 14:01:13 WARN DFSPropertiesConfiguration: Cannot find HUDI_CONF_DIR, please set it as the dir of hudi-defaults.conf
23/10/29 14:01:13 WARN DFSPropertiesConfiguration: Properties file file:/etc/hudi/conf/hudi-defaults.conf not found. Ignoring to load props file
23/10/29 14:01:16 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-hbase.properties,hadoop-metrics2.properties
23/10/29 14:01:17 WARN TaskSetManager: Stage 12 contains a task of very large size (1197 KiB). The maximum recommended task size is 1000 KiB.
# WARNING: Unable to get Instrumentation. Dynamic Attach failed. You may add this JAR as -javaagent manually, or supply -Djdk.attach.allowAttachSelf # WARNING: Unable to attach Serviceability Agent. Unable to attach even with module exceptions: [org.apache.hudi.org.openjdk.jol.vm.sa.SASupportException: Sense failed., org.apache.hudi.org.openjdk.jol.vm.sa.SASupportException: Sense failed., org.apache.hudi.org.openjdk.jol.vm.sa.SASupportException: Sense failed.]
23/10/29 14:01:56 WARN TaskSetManager: Stage 69 contains a task of very large size (1197 KiB). The maximum recommended task size is 1000 KiB.
23/10/29 14:02:19 WARN SparkMetadataTableRecordIndex: Record index not initialized so falling back to GLOBAL_SIMPLE for tagging records
23/10/29 14:02:19 WARN TaskSetManager: Stage 121 contains a task of very large size (1197 KiB). The maximum recommended task size is 1000 KiB.
23/10/29 14:02:39 WARN DAGScheduler: Broadcasting large task binary with size 1123.3 KiB
23/10/29 14:02:41 WARN DAGScheduler: Broadcasting large task binary with size 1223.6 KiB
23/10/29 14:03:51 WARN DAGScheduler: Broadcasting large task binary with size 1134.9 KiB
23/10/29 14:03:52 WARN DAGScheduler: Broadcasting large task binary with size 1235.1 KiB
23/10/29 14:04:52 WARN DAGScheduler: Broadcasting large task binary with size 1137.3 KiB
23/10/29 14:04:54 WARN DAGScheduler: Broadcasting large task binary with size 1237.5 KiB
Comparative Performance¶
Total Upsert Time:¶
- Global Bloom Index: 550.18
- Global Simple Index: 403.153
- Record Level Index: 301.28
Now, let's calculate the percentage by which the Record Level Index (RLI) is faster than the other indexing methods for upsert operations:
- RLI is approximately 45.23% faster than the Global Bloom Index.
- RLI is approximately 25.19% faster than the Global Simple Index.
