Java Apache Spark flatMaps & Обработка данных

Question

Я должен повернуть данные в файл, а затем сохранить его в другом файле.У меня возникли некоторые трудности с поворотом данных.

У меня есть несколько файлов, которые содержат данные, которые выглядят примерно так, как показано ниже.Столбцы имеют переменную длину.Сначала я пытаюсь объединить файлы.Но по какой-то причине вывод не правильный.Я даже не пробовал метод pivot, но не уверен, как его использовать.

Как этого достичь?

File 1:
0,26,27,30,120
201008,100,1000,10,400
201009,200,2000,20,500
201010,300,3000,30,600

File 2:
0,26,27,30,120,145
201008,100,1000,10,400,200
201009,200,2000,20,500,100
201010,300,3000,30,600,150

File 3:
0,26,27,120,145
201008,100,10,400,200
201009,200,20,500,100
201010,300,30,600,150

Output:
201008,26,100
201008,27,1000
201008,30,10
201008,120,400
201008,145,200
201009,26,200
201009,27,2000
201009,30,20
201009,120,500
201009,145,100
.....

Я не совсем знаком со Spark, ноя пытаюсь использовать flatMap и flatMapValues.Я не уверен, как я могу использовать это сейчас, но был бы признателен за некоторые рекомендации.

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.SparkSession;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ExecutionTest {
    public static void main(String[] args) {
        Logger.getLogger("org.apache").setLevel(Level.WARN);
        Logger.getLogger("org.spark_project").setLevel(Level.WARN);
        Logger.getLogger("io.netty").setLevel(Level.WARN);
        log.info("Starting...");
        // Step 1: Create a SparkContext.
        boolean isRunLocally = Boolean.valueOf(args[0]);
        String filePath = args[1];

        SparkConf conf = new SparkConf().setAppName("Variable File").set("serializer",
                "org.apache.spark.serializer.KryoSerializer");
        if (isRunLocally) {
            log.info("System is running in local mode");
            conf.setMaster("local[*]").set("spark.executor.memory", "2g");
        }

        SparkSession session = SparkSession.builder().config(conf).getOrCreate();

        JavaSparkContext jsc = new JavaSparkContext(session.sparkContext());
        jsc.textFile(filePath, 2)
           .map(new Function<String, String[]>() {
                private static final long serialVersionUID = 1L;
                @Override
                public String[] call(String v1) throws Exception {
                    return StringUtils.split(v1, ",");
                }                       
            })
            .foreach(new VoidFunction<String[]>() {
                private static final long serialVersionUID = 1L;
                @Override
                public void call(String[] t) throws Exception {
                    for (String string : t) {
                        log.info(string);
                    }   
                }

            });     
    }

}

Answer 1

Решение в Scala, так как я не человек JAVA, вы должны быть в состоянии адаптироваться. И добавить сортировку, кеш и т. Д.

Данные выглядят следующим образом, 3 файла с повторяющимися записями, избавьтесь от них, если не хотите.

      0, 5,10, 15  20
 202008, 5,10, 15, 20
 202009,10,20,100,200

8 строк, сгенерированных выше.

      0,888,999
 202008,  5, 10
 202009, 10, 20

4 строки, сгенерированные выше.

     0, 5
202009,10

1 строка, которая является дубликатом.

// Bit lazy with columns names, but anyway.

import org.apache.spark.sql.functions.input_file_name  
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._ 
import spark.implicits._

val inputPath: String = "/FileStore/tables/g*.txt"  
val rdd = spark.read.text(inputPath)
  .select(input_file_name, $"value")   
  .as[(String, String)] 
  .rdd

val rdd2 = rdd.zipWithIndex
val rdd3 = rdd2.map(x => (x._1._1, x._2, x._1._2.split(",").toList.map(_.toInt)))  
val rdd4 = rdd3.map { case (pfx, pfx2, list) => (pfx,pfx2,list.zipWithIndex) }
val df = rdd4.toDF()
df.show(false)
df.printSchema()

val df2 = df.withColumn("rankF", row_number().over(Window.partitionBy($"_1").orderBy($"_2".asc)))
df2.show(false)
df2.printSchema()

val df3 = df2.withColumn("elements", explode($"_3"))
df3.show(false)
df3.printSchema()

val df4 = df3.select($"_1", $"rankF", $"elements".getField("_1"), $"elements".getField("_2")).toDF("fn", "line_num", "val", "col_pos")
df4.show(false)
df4.printSchema()

df4.createOrReplaceTempView("df4temp")

val df51 = spark.sql("""SELECT hdr.fn, hdr.line_num, hdr.val AS pfx, hdr.col_pos
                          FROM df4temp hdr
                         WHERE hdr.line_num <> 1
                           AND hdr.col_pos = 0
                     """) 
df51.show(100,false)

val df52 = spark.sql("""SELECT t1.fn, t1.val AS val1, t1.col_pos, t2.line_num, t2.val AS val2
                          FROM df4temp t1, df4temp t2
                         WHERE t1.col_pos  <> 0
                           AND t1.col_pos  = t2.col_pos 
                           AND t1.line_num <> t2.line_num
                           AND t1.line_num = 1
                           AND t1.fn       = t2.fn
                     """)
df52.show(100,false)

df51.createOrReplaceTempView("df51temp")
df52.createOrReplaceTempView("df52temp")

val df53 = spark.sql("""SELECT DISTINCT t1.pfx, t2.val1, t2.val2
                          FROM df51temp t1, df52temp t2
                         WHERE t1.fn = t2.fn 
                           AND t1.line_num = t2.line_num
                     """)
df53.show(false)

возвращается:

+------+----+----+
|pfx   |val1|val2|
+------+----+----+
|202008|888 |5   |
|202009|999 |20  |
|202009|20  |200 |
|202008|5   |5   |
|202008|10  |10  |
|202009|888 |10  |
|202008|15  |15  |
|202009|5   |10  |
|202009|10  |20  |
|202009|15  |100 |
|202008|20  |20  |
|202008|999 |10  |
+------+----+----+ 

Что мы видим, так это Data Wrangling, требующий массированных данных для создания временных представлений и соответствующего соединения с SQL.

Ключевым моментом здесь является умение массировать данные, чтобы упростить задачу. Примечание: нет groupBy и т. Д. Для каждого файла, с различной длиной, присоединение, которое не предпринималось в RDD, слишком негибко. Ранг показывает строку #, поэтому вы знаете первую строку с нулем.

Это то, что мы называем Data Wrangling. Это то, что мы также называем тяжелой работой за несколько пунктов о SO. Это одна из моих лучших работ, а также одна из последних таких попыток.

Слабость решения - это много работы для получения 1-ой записи файла, есть альтернативы. https://www.cyberciti.biz/faq/unix-linux-display-first-line-of-file/ предпроцессинг - это то, что я реально рассмотрю.