用java实现flink中json的聚合

6psbrbz9 于 2021-06-24 发布在 Flink

关注(0)|答案(2)|浏览(367)

我正在从一个文件中读取json数据。
样本数据

{"name":"user1","myparam0":false,"myparam1":"44750004-23df-4960-88be-ba0884291597","myparam2":"36A3BF29-23df-EE2A-76B9-19BC1C854BA7","myparam3":"http://www.seloger.com/","myparam4":"http://www.seloger.com/erreur-temporaire/husk-pie","ver":"4.0.0"}
{"name":"user1","myparam0":true,"myparam1":"44750004-8bff-4960-88be-ba0884291597","myparam2":"36A3BF29-88be-EE2A-76B9-19BC1C854BA7","myparam3":"","myparam4":"http://www.seloger.com/erreur-temporaire/binde","ver":"4.0.0"}

我已经编写了一个示例代码来读取文件并将数据转换成json，如下所示

DataStream<Object> input = env.readTextFile("file:///home/ravisankar/workspace/temporary/input.file")
                .map((line) -> {
                    return JSON.parseFull(line);
                });

现在我要计算一下 myparam3 根据名字15秒后为空。和分组依据 myparam4 ```
Ex: {
"user1": {
"myparams3": 1,
"myparam4": {
"http://www.seloger.com/erreur-temporaire/binde": 1,
"http://www.seloger.com/erreur-temporaire/husk-pie": 1
}
}
}

有没有可能从Flink那里提取这样的数据？？我没有看到任何使用java处理json的例子。谢谢你的时间

Java JSON apache-flink flink-streaming

来源：https://stackoverflow.com/questions/50324201/aggregation-on-json-in-flink-using-java

2条答案

按热度按时间

eivgtgni1#

您可以使用jackson将json解析为一个对象，然后创建一个循环来计算您的元素

private ObjectMapper objectMapper = new ObjectMapper() ;
...
Object element = objectMapper.readValue( jsonString , Object.class );

或者可以使用与“myparam3”：“”匹配的正则表达式并计算匹配项

public static void main( String[] args ) throws IOException
{
    String str = "{\"name\":\"user1\",\"myparam0\":false,\"myparam1\":\"44750004-23df-4960-88be-ba0884291597\",\"myparam2\":\"36A3BF29-23df-EE2A-76B9-19BC1C854BA7\",\"myparam3\":\"http://www.seloger.com/\",\"myparam4\":\"http://www.seloger.com/erreur-temporaire/husk-pie\",\"ver\":\"4.0.0\"}\r\n" + 
            "{\"name\":\"user1\",\"myparam0\":true,\"myparam1\":\"44750004-8bff-4960-88be-ba0884291597\",\"myparam2\":\"36A3BF29-88be-EE2A-76B9-19BC1C854BA7\",\"myparam3\":\"\",\"myparam3\":\"\",\"myparam3\":\"\"\"myparam4\":\"http://www.seloger.com/erreur-temporaire/binde\",\"ver\":\"4.0.0\"}";

    Pattern pattern = Pattern.compile("\"myparam3\":\"\"");

    Matcher matcher = pattern.matcher(str);

    int count = 0;
    while (matcher.find()) {
        count++;
    }
    System.out.println("Matches found : " + count );
}

赞(0）回复(0）举报 2021-06-24

oxf4rvwz2#

您可以将json字符串解析为对象，即通过jackson库，并像往常一样操作java对象流

赞(0）回复(0）举报 2021-06-24