在 Dataflow 管道中解析 Stackdriver LogEntry JSON答案

【问题标题】：Parse Stackdriver LogEntry JSON in Dataflow pipeline在 Dataflow 管道中解析 Stackdriver LogEntry JSON
【发布时间】：2017-10-09 09:01:55
【问题描述】：

我正在构建一个 Dataflow 管道来处理 Stackdriver 日志，从 Pub/Sub 读取数据并将结果写入 BigQuery。当我从 Pub/Sub 读取时，我得到了 LogEntry 对象的 JSON 字符串，但我真正感兴趣的是包含用户日志消息的 protoPayload.line 记录。为了得到那些我需要解析 LogEntry JSON 对象，我发现了一个两年前的 Google example 怎么做：

import com.google.api.client.json.JsonParser;
import com.google.api.client.json.jackson2.JacksonFactory;
import com.google.api.services.logging.model.LogEntry;

try {
    JsonParser parser = new JacksonFactory().createJsonParser(entry);
    LogEntry logEntry = parser.parse(LogEntry.class);
    logString = logEntry.getTextPayload();
}
catch (IOException e) {
    LOG.error("IOException parsing entry: " + e.getMessage());
}
catch(NullPointerException e) {
    LOG.error("NullPointerException parsing entry: " + e.getMessage());
}

不幸的是，这对我不起作用，logEntry.getTextPayload() 返回null。我什至不确定它是否应该工作，因为 Google Cloud 文档中的任何地方都没有提到 com.google.api.services.logging 库，当前的日志库似乎是 google-cloud-logging。

那么，如果有人能建议解析LogEntry 对象的正确或最简单的方法是什么？

【问题讨论】：

标签： google-cloud-dataflow google-cloud-logging google-cloud-stackdriver

【解决方案1】：

我最终使用 gson 库手动解析 LogEntry JSON，特别是使用树遍历方法。这是一个小sn-p：

static class ProcessLogMessages extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) {
        String entry = c.element();

        JsonParser parser = new JsonParser();
        JsonElement element = parser.parse(entry);
        if (element.isJsonNull()) {
            return;
        }
        JsonObject root = element.getAsJsonObject();
        JsonArray lines = root.get("protoPayload").getAsJsonObject().get("line").getAsJsonArray();
        for (int i = 0; i < lines.size(); i++) {
            JsonObject line = lines.get(i).getAsJsonObject();
            String logMessage = line.get("logMessage").getAsString();

            // Do what you need with the logMessage here
            c.output(logMessage);
        }
    }
}

这很简单，对我来说效果很好，因为我只对 protoPayload.line.logMessage 对象感兴趣。但我想如果您需要处理许多属性，这不是解析 LogEntry 对象的理想方法。

【讨论】：