scala - 如何使用 Spark 读取不断更新的 HDFS 目录并根据字符串(行)将输出拆分为多个 HDFS 文件？

Question

详细场景 -> HDFS 目录，其中“提供”了多种类型的银行账户事件的新日志数据。每行代表一个随机事件类型，每行 (String) 包含文本“ActivityType= <TheTypeHere>”。

在 Spark-Scala 中，读取 HDFS 目录中的输入文件并输出多个 HDFS 文件的最佳方法是什么，其中每个 ActivityType 都写入其自己的新文件？

Answer 1

改编为声明的第一个答案:

The location of the "key" string is random within the parent String, the only thing that is guaranteed is that it contains that sub-string, in this case "ActivityType" followed by some val.

问题真的是关于这个的。开始了:

// SO Question
val rdd = sc.textFile("/FileStore/tables/activitySO.txt")  
val rdd2 = rdd.map(x => (x.slice (x.indexOfSlice("ActivityType=<")+14, x.indexOfSlice(">", (x.indexOfSlice("ActivityType=<")+14))), x))
val df = rdd2.toDF("K", "V")
df.write.partitionBy("K").text("SO_QUESTION2")

输入是:

ActivityType=<ACT_001>,34,56,67,89,90
3,4,4,ActivityType=<ACT_002>,A,1,2
ABC,ActivityType=<ACT_0033>
DEF,ActivityType=<ACT_0033>

输出是 3 个文件，其中 key 是例如不是 ActivityType=，而是 ACT_001 等。关键数据没有被剥离，它仍然存在于字符串中。如果需要，您可以修改它以及输出位置和格式。