使用 Spark 进行字符串转换

Question

我正在学习 Spark，并尝试编写非常简单的应用程序。 作为输入，我有日志字符串，看起来像

INFO - {timestamp} - {path} - {json message}
INFO - 124534234534534 - test.class - {"message": "something happened"]

我想将它传递给 ElasticSearch。所以我需要获取 {timestamp} 并将其放入 {json message} 的新字段，所以它应该看起来像

{"timestamp": "1234343132", "message": "something happened"}

有人可以使用 Java 帮助我进行此转换吗？

Answer 1

1. 创建一个Function<String, String>，其中包含一行日志和returns JSON 字符串。

   Function<String, String> f = new Function<String, String>() {
     public String call(String s) { return ...; }
   }
   

2. 使用SparkContext.textFile

   读取数据

   JavaSparkContext sc = ...;
   JavaRDD<String> rdd = sc.textFile(...)
   

3. map 使用第 1 点中定义的函数创建了 RDD。

   rdd.map(f);