update notes

Author: LorettaYoo

01-HelloSpark/HelloSpark.py

@@ -23,7 +23,8 @@
     survey_raw_df = load_survey_df(spark, sys.argv[1])
     partitioned_survey_df = survey_raw_df.repartition(2)
     count_df = count_by_country(partitioned_survey_df)
-    count_df.show()
+    count_df.collect()
 
-    logger.info("Finished HelloSpark")
-    spark.stop()
+    input("Please Print Enter")
+    # logger.info("Finished HelloSpark")
+    # spark.stop()

03-HelloSparkSQL/HelloSparkSQL.py

@@ -22,7 +22,7 @@
         .option("inferSchema", "true") \
         .csv(sys.argv[1])
 
-    surveyDF.createOrReplaceTempView("survey_tbl")
+    surveyDF.createOrReplaceTempView("survey_tbl")  # 创建表
     countDF = spark.sql("select Country, count(1) as count from survey_tbl where Age<40 group by Country")
 
     countDF.show()

05-DataSinkDemo/DataSinkDemo.py

@@ -23,11 +23,11 @@
     logger.info("Num Partitions after: " + str(partitionedDF.rdd.getNumPartitions()))
     partitionedDF.groupBy(spark_partition_id()).count().show()
 
-    partitionedDF.write \
-        .format("avro") \
-        .mode("overwrite") \
-        .option("path", "dataSink/avro/") \
-        .save()
+    # partitionedDF.write \
+    #     .format("avro") \
+    #     .mode("overwrite") \
+    #     .option("path", "dataSink/avro/") \
+    #     .save()
 
     flightTimeParquetDF.write \
         .format("json") \

06-SparkSQLTableDemo/SparkSQLTableDemo.py

@@ -19,8 +19,16 @@
     spark.sql("CREATE DATABASE IF NOT EXISTS AIRLINE_DB")
     spark.catalog.setCurrentDatabase("AIRLINE_DB")
 
+    # flightTimeParquetDF.write \
+    #     .mode("overwrite") \
+    #     .partitionBy("ORIGIN", "OP_CARRIER") \
+    #     .saveAsTable("flight_data_tbl")
+    #
     flightTimeParquetDF.write \
+        .format("csv") \
         .mode("overwrite") \
+        .bucketBy(5, "OP_CARRIER", "ORIGIN") \
+        .sortBy("OP_CARRIER", "ORIGIN") \
         .saveAsTable("flight_data_tbl")
 
-    logger.info(spark.catalog.listTables("AIRLINE_DB"))
+logger.info(spark.catalog.listTables("AIRLINE_DB"))

11-UDFDemo/UDFDemo.py

@@ -33,14 +33,21 @@ def parse_gender(gender):
 
     survey_df.show(10)
 
+    # 注册 UDF 方法1 在Dataframe列对象表达式中使用你的函数
     parse_gender_udf = udf(parse_gender, returnType=StringType())
+
+    # UDF（）函数并不创建目录条目｡ 因此, 我们不应该得到任何输出｡
     logger.info("Catalog Entry:")
     [logger.info(r) for r in spark.catalog.listFunctions() if "parse_gender" in r.name]
 
+    # 我们使用了withColumn（）转换, 它只影响Gender字段｡
     survey_df2 = survey_df.withColumn("Gender", parse_gender_udf("Gender"))
     survey_df2.show(10)
 
+    # 注册 UDF 方法2 将其注册为SQL 函数 在SQL表达式中使用你的函数
     spark.udf.register("parse_gender_udf", parse_gender, StringType())
+
+    # 找到一个条目并打印函数的详细信息
     logger.info("Catalog Entry:")
     [logger.info(r) for r in spark.catalog.listFunctions() if "parse_gender" in r.name]
 

12-MiscDemo/MiscDemo.py

@@ -19,20 +19,23 @@
                  ("Rosy", "7", "8", "63"),  # 1963
                  ("Abdul", "23", "5", "81")]  # 1981
 
+    # createDataFrame 新建
     raw_df = spark.createDataFrame(data_list).toDF("name", "day", "month", "year").repartition(3)
     raw_df.printSchema()
 
+    # know the use of cast()
     final_df = raw_df.withColumn("id", monotonically_increasing_id()) \
         .withColumn("day", col("day").cast(IntegerType())) \
         .withColumn("month", col("month").cast(IntegerType())) \
         .withColumn("year", col("year").cast(IntegerType())) \
-        .withColumn("year", when(col("year") < 20, col("year") + 2000)
+        .withColumn("year", \
+                    when(col("year") < 20, col("year") + 2000)
                     .when(col("year") < 100, col("year") + 1900)
                     .otherwise(col("year"))) \
         .withColumn("dob", expr("to_date(concat(day,'/',month,'/',year), 'd/M/y')")) \
         .drop("day", "month", "year") \
         .dropDuplicates(["name", "dob"]) \
-        # .sort(expr("dob desc")) This doesn't seem to be working
         .sort(col("dob").desc())
+    # .sort(expr("dob desc")) This doesn't seem to be working
 
-    final_df.show()
+final_df.show()

13-AggDemo/AggDemo.py

@@ -31,6 +31,7 @@
         "avg(UnitPrice) as AvgPrice"
     ).show()
 
+    # 首先创建表，再用 spark.sql
     invoice_df.createOrReplaceTempView("sales")
     summary_sql = spark.sql("""
           SELECT Country, InvoiceNo,

16-RankingDemo/RankingDemo.py

@@ -20,6 +20,7 @@
         .orderBy(f.col("InvoiceValue").desc()) \
         .rowsBetween(Window.unboundedPreceding, Window.currentRow)
 
+    # 找出每个国家InvoiceValue 的第一名
     df = summary_df.withColumn("Rank", f.dense_rank().over(rank_window)) \
         .where(f.col("Rank") == 1) \
         .sort("Country", "WeekNumber") \

19-ShuffleJoinDemo/SuffleJoinDemo.py

@@ -1,4 +1,5 @@
 from pyspark.sql import SparkSession
+from pyspark.sql.functions import broadcast
 
 from lib.logger import Log4j
 
@@ -14,10 +15,12 @@
     flight_time_df1 = spark.read.json("data/d1/")
     flight_time_df2 = spark.read.json("data/d2/")
 
+    # 这种配置将确保我们在shuffle之后得到三个分区, 这意味着有三个reduce交换｡
     spark.conf.set("spark.sql.shuffle.partitions", 3)
 
     join_expr = flight_time_df1.id == flight_time_df2.id
-    join_df = flight_time_df1.join(flight_time_df2, join_expr, "inner")
+    # join_df = flight_time_df1.join(flight_time_df2, join_expr, "inner")
+    join_df = flight_time_df1.join(broadcast(flight_time_df2), join_expr, "inner")
 
     join_df.collect()
-    input("press a key to stop...")
+    input("press a key to stop...")