hudi-dynamodb

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 changeme

Backend¶


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]:

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)

Pythonist

Saturday, October 28, 2023

How to Unlock Data Insights from Hudi Metrics for Your Data Lake using Elastic Search and Kibana

How to Unlock Data Insights from Hudi Metrics for Your Data Lake using Elastic Search and Kibana¶

Setup Elastic Stack Locally¶

Backend¶

Step 1: Define imports¶

Step 2: Create Spark Session¶

Hudi Settings¶

Data Generator Class¶

Inserting Documents into datalake¶

100 samples¶

1000 samples¶

10000 samples¶

100000 samples¶

Delete Workloads¶

No comments:

Post a Comment

Let's Migrate Iceberg Tables from One Catalog to Another | Simple Hands-on Lab