0


Flink实践代码-TableAPI 与 DataStream 互转

1.代码与含义解释

1.1 思路

Flink 获取数据流后,需要做数据过滤那么首先就要有一下几个步骤:

  1. 构建运行环境
  2. 接入数据流
  3. TableAPI 与 DataStream 互转,实现 SQL 查询
1.2 直接上代码
package com.youtree.flink;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
 
public class Visit_Info_for_table {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);

        DataStream<String> VisitInfo = env.readTextFile("FilePathinfos");

        DataStream<JSONObject> jsonObjectDataStream = VisitInfo.flatMap(new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(s);
                        jsonObject.remove("visit_info");
                        collector.collect(jsonObject);
                    }
                })
                .filter(value -> String.valueOf(value.get("is_valid")).equals("1"))
                .filter(Objects::nonNull);

        DataStream<Row> rowDataStream = jsonObjectDataStream.flatMap(new FlatMapFunction<JSO
标签: flink 大数据

本文转载自: https://blog.csdn.net/danevc/article/details/137152165
版权归原作者 danevc 所有, 如有侵权,请联系我们删除。

“Flink实践代码-TableAPI 与 DataStream 互转”的评论:

还没有评论