How to Unlock Data Insights from Hudi Metrics for Your Data Lake using Elastic Search and Kibana¶
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-d59215b8-7f92-4d80-989d-c6a324727973;1.0 confs: [default] found org.apache.hudi#hudi-spark3.4-bundle_2.12;0.14.0 in central :: resolution report :: resolve 211ms :: 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-d59215b8-7f92-4d80-989d-c6a324727973 confs: [default] 0 artifacts copied, 1 already retrieved (0kB/10ms) 23/10/28 13:57:25 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).
Hudi Settings¶
In [3]:
db_name = "hudidb"
table_name = "employees"
path = f"file:///Users/soumilnitinshah/Downloads/{db_name}/{table_name}" # Updated path
method = 'upsert'
table_type = "COPY_ON_WRITE"
recordkey = "emp_id"
precombine = "emp_id"
lambda_function_url = "http://127.0.0.1:5000"
partition_fields = "state"
hudi_options = {
'hoodie.table.name': table_name,
'hoodie.datasource.write.table.type': table_type,
'hoodie.datasource.write.table.name': table_name,
'hoodie.datasource.write.operation': method,
'hoodie.datasource.write.recordkey.field': recordkey,
'hoodie.datasource.write.precombine.field': precombine,
"hoodie.datasource.write.partitionpath.field": partition_fields
,"hoodie.write.commit.callback.http.url":lambda_function_url
,"hoodie.write.commit.callback.on":"true"
,"hoodie.write.commit.callback.http.timeout.seconds":"180"
}
Data Generator Class¶
In [4]:
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)
]
Inserting Documents into datalake¶
100 samples¶
In [6]:
data = DataGenerator.get_data(samples=100)
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.write.format("hudi"). \
options(**hudi_options). \
mode("append"). \
save(path)
1000 samples¶
In [7]:
data = DataGenerator.get_data(samples=1000)
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.write.format("hudi"). \
options(**hudi_options). \
mode("append"). \
save(path)
10000 samples¶
In [8]:
data = DataGenerator.get_data(samples=10000)
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.write.format("hudi"). \
options(**hudi_options). \
mode("append"). \
save(path)
100000 samples¶
In [30]:
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.write.format("hudi"). \
options(**hudi_options). \
mode("append"). \
save(path)
23/10/28 12:57:56 WARN TaskSetManager: Stage 611 contains a task of very large size (1196 KiB). The maximum recommended task size is 1000 KiB.
Delete Workloads¶
In [35]:
path = 'file:///Users/soumilnitinshah/Downloads/hudidb/employees'
spark.read.format("hudi").load(path).createOrReplaceTempView("hudi_snapshot")
spark.sql("select * from hudi_snapshot where salary >= 50000 ").count()
Out[35]:
79616
In [36]:
path = 'file:///Users/soumilnitinshah/Downloads/hudidb/employees'
spark.read.format("hudi").load(path).createOrReplaceTempView("hudi_snapshot")
delete_df = spark.sql("select * from hudi_snapshot where salary >= 50000 ")
db_name = "hudidb"
table_name = "employees"
method = 'delete'
table_type = "COPY_ON_WRITE"
recordkey = "emp_id"
precombine = "emp_id"
lambda_function_url = "http://127.0.0.1:5000"
partition_fields = "state"
hudi_options = {
'hoodie.table.name': table_name,
'hoodie.datasource.write.table.type': table_type,
'hoodie.datasource.write.table.name': table_name,
'hoodie.datasource.write.operation': method,
'hoodie.datasource.write.recordkey.field': recordkey,
'hoodie.datasource.write.precombine.field': precombine,
"hoodie.datasource.write.partitionpath.field": partition_fields
,"hoodie.write.commit.callback.http.url":lambda_function_url
,"hoodie.write.commit.callback.on":"true"
,"hoodie.write.commit.callback.http.timeout.seconds":"180"
}
delete_df.write.format("hudi"). \
options(**hudi_options). \
mode("append"). \
save(path)
23/10/28 13:14:05 WARN Executor: Managed memory leak detected; size = 5269600 bytes, task 0.0 in stage 659.0 (TID 3842)
No comments:
Post a Comment