1.代码与含义解释
1.1 思路
Flink 获取数据流后,需要做数据过滤那么首先就要有一下几个步骤:文章来源地址https://www.toymoban.com/news/detail-847540.html
- 构建运行环境
- 接入数据流
- 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
文章来源:https://www.toymoban.com/news/detail-847540.html
到了这里,关于Flink实践代码-TableAPI 与 DataStream 互转的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!