使用 Java 中的 Apache Spark 使用数组解析 Json 对象并映射到多对答案

【问题标题】：Parse Json Object with an Array and Map to Multiple Pairs with Apache Spark in Java使用 Java 中的 Apache Spark 使用数组解析 Json 对象并映射到多对
【发布时间】：2016-11-15 16:15:29
【问题描述】：

我用谷歌搜索了一整天，找不到直接的答案，所以最后在这里发布了一个问题。

我有一个包含行分隔的 json 对象的文件：

{"device_id": "103b", "timestamp": 1436941050, "rooms": ["Office", "Foyer"]}
{"device_id": "103b", "timestamp": 1435677490, "rooms": ["Office", "Lab"]}
{"device_id": "103b", "timestamp": 1436673850, "rooms": ["Office", "Foyer"]}

我的目标是用 Java 中的 Apache Spark 解析这个文件。我引用了How to Parsing CSV or JSON File with Apache Spark，到目前为止，我可以使用Gson成功地将每一行json解析为JavaRDD。

JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> data = sc.textFile("fileName");
JavaRDD<JsonObject> records = data.map(new Function<String, JsonObject>() {
    public JsonObject call(String line) throws Exception {
        Gson gson = new Gson();
        JsonObject json = gson.fromJson(line, JsonObject.class);
        return json;
    }
});

我真正陷入困境的地方是我想反序列化“房间”数组，以便它适合我的班级事件。

public class Event implements Serializable {
    public static final long serialVersionUID = 42L;
    private String deviceId;
    private int timestamp;
    private String room;
    // constructor , getters and setters 
}

换句话说，从这一行开始：

{"device_id": "103b", "timestamp": 1436941050, "rooms": ["Office", "Foyer"]}

我想在 Spark 中创建两个 Event 对象：

obj1: deviceId = "103b", timestamp = 1436941050, room = "Office"
obj2: deviceId = "103b", timestamp = 1436941050, room = "Foyer"

我做了我的小搜索并尝试了 flatMapVue，但没有运气...它给我一个错误...

JavaRDD<Event> events = records.flatMapValue(new Function<JsonObject, Iterable<Event>>() {
    public Iterable<Event> call(JsonObject json) throws Exception {
        JsonArray rooms = json.get("rooms").getAsJsonArray();
        List<Event> data = new LinkedList<Event>();
        for (JsonElement room : rooms) {
            data.add(new Event(json.get("device_id").getAsString(), json.get("timestamp").getAsInt(), room.toString()));
        }
        return data;
    }
});

我对 Spark 和 Map/Reduce 非常陌生。如果您能帮助我，我将不胜感激。提前致谢！

【问题讨论】：

请发布您的错误。编辑您的帖子并添加堆栈跟踪

标签： java json apache-spark

【解决方案1】：

如果将 json 数据加载到 DataFrame:

DataFrame df = sqlContext.read().json("/path/to/json");

explode 可以轻松做到这一点。

df.select(
    df.col("device_id"),
    df.col("timestamp"),
    org.apache.spark.sql.functions.explode(df.col("rooms")).as("room")
);

输入：

{"device_id": "1", "timestamp": 1436941050, "rooms": ["Office", "Foyer"]}
{"device_id": "2", "timestamp": 1435677490, "rooms": ["Office", "Lab"]}
{"device_id": "3", "timestamp": 1436673850, "rooms": ["Office", "Foyer"]}

你会得到：

+---------+------+----------+
|device_id|  room| timestamp|
+---------+------+----------+
|        1|Office|1436941050|
|        1| Foyer|1436941050|
|        2|Office|1435677490|
|        2|   Lab|1435677490|
|        3|Office|1436673850|
|        3| Foyer|1436673850|
+---------+------+----------+

【讨论】：

感谢您让我知道这个有用的功能。我不知道 Spark 像 UDF 一样支持 Hive。很有帮助！
spark与hive完全兼容(*≧▽≦)

【解决方案2】：

val formatrecord = records.map(fromJson[mapClass](_))

mapClass 应该是用于映射记录 json 中的对象的案例类。

【讨论】：