0

我有一个 RDD 转换为以下结构的数据帧：

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

Key 是一个四舍五入到一小时的时间戳，value 是一个 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
...

仅存储实际的 rddeventsRdd.saveAsTextFile("/datalake");并不能解决问题，因为所有事件最终都在一个文件中。此外，该文件在下一个“回合”中被覆盖。

那么我该怎么做呢？我读了一些关于分区的文章，但它们并没有真正帮助。我实际上正在考虑切换到 Kafka Connect 而根本不使用 Spark。

下面是我尝试存储事件的一些代码（目前仅在本地 fs 上）

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;
    }
4

1 回答 1

0

我们通过使用 Kafka Connect HDFS 连接器并提供自定义序列化器类来将 Protobuf 消息从 Kafka 转换为 JSON 来完成它。

于 2019-05-03T09:19:32.313 回答