使用Spark进行字符串转换

Question

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

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

我想将它传递给ElasticSearch。所以，我需要{}时间戳并把它新的领域，以{JSON消息}，所以它应该像

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

有人可以帮助我使用Java这种转变？

Answer 1

1. 创建Function<String, String>这需要一个行日志，并返回JSON字符串。使用SparkContext.textFile

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

2. 使用函数创建RDD map

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

3. 读取数据点定义1

   rdd.map(f);