как заменить отличный () на reducebykey - PullRequest
Новая
       8

как заменить отличный () на reducebykey

1 голос
/ 11 июля 2020

У меня есть сценарий, в котором приведенный ниже код в целом занимает более 10 часов для> 2 миллиардов записей. даже я пробовал с 35 экземплярами кластера i3, но все равно производительность была плохой. Я ищу возможность заменить отличную () на reduceByKey () и получить предложение по повышению производительности ... 

    val df = spark.read.parquet(out)
      
     val df1 = df.
    select($"ID", $"col2", $"suffix",
   $"date", $"year", $"codes")

   val df2 = df1.
    repartition(
      List(col("ID"), col("col2"), col("suffix"), col("date"),
        col("year"), col("codes")): _*
    ).distinct()
      
       val df3 = df2.withColumn("codes", expr("transform(codes, (c,s) -> (d,s) )"))
    
       df3.createOrReplaceTempView("df3")
    
       val df4 = spark.sql(
         """SELECT
               ID, col2, suffix
               d.s as seq,
               d.c as code,
               year,date
               FROM
                df3
                 LATERAL VIEW explode(codes) exploded_table as d
                 """)
    
       df4.
         repartition(
           600,
           List(col("year"), col("date")): _*).
         write.
         mode("overwrite").
         partitionBy("year", "date").
         save(OutDir)

1 Ответ

0 голосов
/ 11 июля 2020

Я думаю, что independent () реализуется с помощью reduceByKey (reduce), но если вы хотите реализовать это самостоятельно, вы можете что-то сделать 

val array=List((1,2),(1,3),(1,5),(1,2),(2,2),(2,2),(3,2),(3,2),(4,1),(1,3))
val pairRDD=session.sparkContext.parallelize(array)
val distinctResult=pairRDD.map(x => (x, null)).reduceByKey((x, _) => x)
...