PySpark aggregations

PySpark परिचय

Benjamin Schmidt

Data Engineer

PySpark SQL aggregations अवलोकन

सामान्य SQL aggregations spark.sql() के साथ काम करती हैं

# SQL aggregation query
spark.sql("""
  SELECT Department, SUM(Salary) AS Total_Salary, AVG(Salary) AS Average_Salary
  FROM employees
  GROUP BY Department
""").show()

DataFrame और SQL ऑपरेशंस को जोड़ना

# Filter salaries over 3000
filtered_df = df.filter(df.Salary > 3000)

# Register filtered DataFrame as a view
filtered_df.createOrReplaceTempView("filtered_employees")


# Aggregate using SQL on the filtered view
spark.sql("""
    SELECT Department, COUNT(*) AS Employee_Count
    FROM filtered_employees
    GROUP BY Department
""").show()

Aggregations में डेटा टाइप संभालना

# Example of type casting
data = [("HR", "3000"), ("IT", "4000"), ("Finance", "3500")]
columns = ["Department", "Salary"]
df = spark.createDataFrame(data, schema=columns)


# Convert Salary column to integer
df = df.withColumn("Salary", df["Salary"].cast("int"))

# Perform aggregation
df.groupBy("Department").sum("Salary").show()

Aggregations के लिए RDDs

# Example of aggregation with RDDs
rdd = df.rdd.map(lambda row: (row["Department"], row["Salary"]))


rdd_aggregated = rdd.reduceByKey(lambda x, y: x + y)


print(rdd_aggregated.collect())

PySpark aggregations के लिए सर्वोत्तम तरीके

पहले फ़िल्टर करें: aggregations से पहले डेटा साइज घटाएँ
डेटा टाइप संभालें: डेटा साफ और सही टाइप में रखें
पूरे डेटासेट पर चलने वाले ऑपरेशंस से बचें: groupBy() जैसे ऑपरेशंस कम करें
सही इंटरफेस चुनें: ऑप्टिमाइज़ेशन के कारण अधिकतर कामों में DataFrames को प्राथमिकता दें
परफॉर्मेंस मॉनिटर करें: execution plan देखने और ट्यून करने के लिए explain() का उपयोग करें

मुख्य बातें

PySpark SQL Aggregations: डेटा सारांशित करने के लिए SUM() और AVERAGE() जैसी functions
DataFrames और SQL: लचीले data manipulation के लिए दोनों तरीकों का संयोजन
Data Types संभालना: aggregations के दौरान type mismatch की समस्याएँ सुलझाना
RDDs बनाम DataFrames: समझौते समझें और सही टूल चुनें

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

PySpark परिचय