Databricks में डेटा ट्रांसफॉर्मेशन

Databricks कॉन्सेप्ट्स

Kevin Barlow

Data Practitioner

डेटा इंजीनियरिंग के लिए SQL

SQL

Database Administrators (DBAs) के लिए परिचित
मानक बदलावों के लिए बढ़िया
प्री-डिफाइंड UDFs चलाएँ

-- Creating a new table in SQL

CREATE TABLE table_name
USING delta
AS (
  SELECT *
  FROM source_table
  WHERE date >= '2023-01-01'
)

डेटा इंजीनियरिंग के लिए अन्य भाषाएँ

Python, R, Scala

सॉफ्टवेयर इंजीनियर्स के लिए परिचित
स्टैंडर्ड और जटिल ट्रांसफॉर्मेशन
कस्टम फंक्शन इस्तेमाल करें और परिभाषित करें

#Creating a new table in Pyspark

spark
  .read
  .table('source_table')
  .filter(col('date') >= '2023-01-01')
  .write
  .saveAsTable('table_name')

कॉमन ट्रांसफॉर्मेशन

स्कीमा मैनिपुलेशन

कॉलम जोड़ें और हटाएँ
कॉलम फिर से परिभाषित करें

#Pyspark

df
  .withColumn(col('newCol'), ...)
  .drop(col('oldCol'))

फ़िल्टरिंग

DataFrame को डेटा के सबसेट तक घटाएँ
एक से अधिक मानदंड पास करें

#Pyspark

df
  .filter(col('date') >= target_date)
  .filter(col('id') IS NOT NULL)

कॉमन ट्रांसफॉर्मेशन (आगे)

नेस्टेड डेटा

Arrays या Struct डेटा
फैलाएँ या समेटें

df
  .explode(col('arrayCol')) #wide to long
  .flatten(col('items')) #long to wide

एग्रीगेशन

कॉलम के आधार पर ग्रुप करें
सार-संक्षेप गणनाएँ करें

df
  .groupBy(col('region'))
  .agg(sum(col('sales')))

Auto Loader

Auto Loader डेटा लेक में आते ही नए डेटा फाइल्स प्रोसेस करता है.

इंक्रीमेंटल प्रोसेसिंग
एफिशिएंट प्रोसेसिंग
ऑटोमैटिक

spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load(file_path)

Auto Loader आरेख

¹ https://www.databricks.com/blog/2020/02/24/introducing-databricks-ingest-easy-data-ingestion-into-delta-lake.html

Structured Streaming

स्ट्रीमिंग पाइपलाइन

spark.readStream
    .format("kafka")
    .option("subscribe", "<topic>")
    .load()
    .join(table_df, 
      on="<id>", how="left")
    .writeStream
    .format("kafka")
    .option("topic", "<topic>")
    .start()

अभ्यास करते हैं!

Databricks कॉन्सेप्ट्स