Flink实践代码-TableAPI 与 DataStream 互转

这篇具有很好参考价值的文章主要介绍了Flink实践代码-TableAPI 与 DataStream 互转。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1.代码与含义解释

1.1 思路

Flink 获取数据流后,需要做数据过滤那么首先就要有一下几个步骤:文章来源地址https://www.toymoban.com/news/detail-847540.html

  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实践代码-TableAPI 与 DataStream 互转的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Flink DataStream之输出数据到File中

    新建类 启动程序并启动nc -lp 输入数据: 正在写入的文件会有inprogress的标识(在指定的目录下生成文件时会按照日期的年月日时进行分目录,因为我在执行时的时间是2023/7/12 22点,所以它就会自动生成一个2023-07-12--22目录,分桶策略也可以自己在代码中配置。):  当满足滚

    2024年02月15日
    浏览(39)
  • Flink 读写MySQL数据(DataStream和Table API)

    Flink提供了基于JDBC的方式,可以将读取到的数据写入到MySQL中;本文通过两种方式将数据下入到MySQL数据库,其他的基于JDBC的数据库类似,另外,Table API方式的Catalog指定为Hive Catalog方式,持久化DDL操作。 另外,JDBC 连接器允许使用 JDBC 驱动程序从任何关系数据库读取数据并将

    2023年04月09日
    浏览(46)
  • Flink TableAPI Window and Watermarket

    本次主要是弄清楚.批流统一 的处理方式,因为它是使用SQL来操作批流计算的.所以它怎么设置算子并行度?如何设置窗口?如何处理流式数据?等等 有很多疑问. 我还是觉得直接使用流计算的API更好.流批一体API最终也是转换成流式计算,最主要的是使用sql来设置算子或者窗口,并不直

    2024年02月12日
    浏览(37)
  • Flink TableAPI Aggregation And DataType

    这里整理下聚合的优化选项 以及 数据类型 SQL 是数据分析中使用最广泛的语言。Flink Table API 和 SQL 使用户能够以更少的时间和精力定义高效的流分析应用程序。此外,Flink Table API 和 SQL 是高效优化过的,它集成了许多查询优化和算子优化。 但并不是所有的优化都是默认开启

    2024年02月16日
    浏览(36)
  • Flink DataStream API CDC同步MySQL数据到StarRocks

    Flink:1.16.1 pom文件如下 Java代码 SourceAndSinkInfo 类,用于定义source和sink的IP、端口、账号、密码信息 DataCenterShine实体类,字段与数据库一一对应。 StarRocksPrimary 实体类 FieldInfo注解类,用于标记字段序号、是否为主键、是否为空,后续生成TableSchema需要使用到。 TableName 注解类,

    2024年02月03日
    浏览(78)
  • 大数据学习之Flink算子、了解DataStream API(基础篇一)

    注: 本文只涉及DataStream 原因:随着大数据和流式计算需求的增长,处理实时数据流变得越来越重要。因此,DataStream由于其处理实时数据流的特性和能力,逐渐替代了DataSet成为了主流的数据处理方式。 目录 DataStream API (基础篇) 前摘: 一、执行环境 1. 创建执行环境 2. 执

    2024年01月23日
    浏览(53)
  • 【Flink系列七】TableAPI和FlinkSQL初体验

    Apache Flink 有两种关系型 API 来做流批统一处理:Table API 和 SQL Table API 是用于 Scala 和 Java 语言的查询API,它可以用一种非常直观的方式来组合使用选取、过滤、join 等关系型算子。  Flink SQL 是基于 Apache Calcite 来实现的标准 SQL。无论输入是连续的(流式)还是有界的(批处理

    2024年02月03日
    浏览(39)
  • 【flink番外篇】19、Datastream数据类型到Table schema映射示例

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年01月21日
    浏览(53)
  • flink cdc数据同步,DataStream方式和SQL方式的简单使用

    目录 一、flink cdc介绍 1、什么是flink cdc 2、flink cdc能用来做什么 3、flink cdc的优点 二、flink cdc基础使用 1、使用flink cdc读取txt文本数据 2、DataStream的使用方式 3、SQL的方式 总结 flink cdc是一个由阿里研发的,一个可以直接从MySQL、PostgreSQL等数据库直接读取全量数据和增量变更数

    2024年02月13日
    浏览(41)
  • 【Apache Flink】Flink DataStream API的基本使用

    Flink DataStream API的基本使用 Flink DataStream API主要用于处理无界和有界数据流 。 无界数据流 是一个持续生成数据的数据源,它没有明确的结束点,例如实时的交易数据或传感器数据。这种类型的数据流需要使用Apache Flink的实时处理功能来连续地处理和分析。 有界数据流 是一个

    2024年02月06日
    浏览(39)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包