Как создать озеро данных из Кафки в Hdfs с помощью Spark - хранение в пользовательских каталогах? - PullRequest
Новая
       69

Как создать озеро данных из Кафки в Hdfs с помощью Spark - хранение в пользовательских каталогах?

0 голосов
/ 25 апреля 2019

У меня есть СДР, преобразованный в dataFrame следующей структуры: 

+-------------+--------------------+
|          key|               value|
+-------------+--------------------+
|1556110998000|{"eventId":"55108...|
|1556110998000|{"eventId":"558ac...|
|1556110998000|{"eventId":"553c0...|
|1556111001600|{"eventId":"56886...|
|1556111001600|{"eventId":"569ad...|
|1556111001600|{"eventId":"56b34...|
|1556110998000|{"eventId":"55d1b...|
...

Ключ - это временная метка, округленная до одного часа, значение - это строка json.

Я хочу сохранить значения в разных сегментах в соответствии с отметкой времени.Поэтому в основном я хочу, чтобы структура была следующей: 

...
/datalake/2019/03/31/03
/datalake/2019/03/31/04
/datalake/2019/03/31/05
...
/datalake/2019/04/25/08
/datalake/2019/04/25/09
...

Сохранение фактического rdd только с помощью eventsRdd.saveAsTextFile("/datalake"); не помогает, так как все события заканчиваются в одном файле.Кроме того, этот файл будет перезаписан в следующем «раунде».

Так как мне поступить?Я прочитал некоторые статьи о разделении, но они действительно не помогли.Я на самом деле думаю о переходе на Kafka Connect и вообще не использую Spark для этого.

Ниже приведен код, который я пытался сохранить события (сейчас только на локальном фс) 

private void saveToDatalake(JavaRDD<E> eventsRdd) {
        JavaPairRDD<Long, String> longEJavaPairRdd = eventsRdd
                .mapToPair(event -> new Tuple2<>(calculateRoundedDownTimestampFromSeconds(event.getTimestamp()), serialize(event)));

        SparkSession sparkSession = SparkSession.builder().appName("Build a DataFrame from Scratch").master("local[*]").getOrCreate();
        StructType dataFrameSchema = DataTypes
                .createStructType(new StructField[]
                        {DataTypes.createStructField("key", DataTypes.LongType, false),
                                DataTypes.createStructField("value", DataTypes.StringType, false),

                        });
        JavaRDD<Row> rowRdd = longEJavaPairRdd.map(pair -> RowFactory.create(pair._1, pair._2));

        Dataset<Row> dataFrame = sparkSession.sqlContext().createDataFrame(rowRdd, dataFrameSchema);

        Dataset<Row> buckets = dataFrame.select("key").dropDuplicates();
        //buckets.show();
        buckets.foreach(bucket -> {
            Dataset<Row> valuesPerBucket = dataFrame.where(dataFrame.col("key").equalTo(bucket)).select("value");
            //valuesPerBucket.show();
            long timestamp = bucket.getLong(0);
            valuesPerBucket.rdd().saveAsTextFile("/data/datalake/" + calculateSubpathFromTimestamp(timestamp));
        });

    }

private String calculateSubpathFromTimestamp(long timestamp) {
        String FORMAT = "yyyy/MM/dd/HH";
        ZoneId zone = ZoneId.systemDefault();
        DateTimeFormatter df = DateTimeFormatter.ofPattern(FORMAT).withZone(zone);
        String time = df.format(Instant.ofEpochMilli(timestamp));
        System.out.println("Formatted Date " + time);
        return time;
    }

1 Ответ

0 голосов
/ 03 мая 2019

Мы сделали это с помощью Kafka Connect HDFS Connector и предоставили собственный класс Serializer для преобразования сообщений Protobuf из Kafka в JSON.

...