Flinksql Kafka 接收流数据并打印到控制台

芒果1年前技术文章1694

本文目的
使用Flink SQL创建一个流处理作业,将来自Kafka主题"dahua_picrecord"的数据写入到另一个表”print_table”控制台中。
使用sql-client前 需要启动yarn-session哦
首先需要在CREATE TABLE
CREATE TABLE test_source (
  objId STRING,
  data STRING,
  capTime STRING,
  dataType STRING,
  channelCode STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'test',
  'properties.bootstrap.servers' = '172.16.121.194:9092',
  'properties.group.id' = 'test-dataq-01',
  'format' = 'json',
  'scan.startup.mode' = 'earliest-offset'
);

CFC9E3F3-A2FA-43C5-A99C-C765F1A0ACAB.png
创建”print_table"
CREATE TABLE print_table (
  objId STRING,
  data STRING,
  capTime STRING,
  dataType STRING,
  channelCode STRING
) WITH (
  'connector' = 'print'
);
ACF330CB-6770-4C05-8D86-23C622FAD014.png
将数据从test_source 插入到 print_table 中
INSERT INTO print_table
SELECT objId, data, capTime, dataType, channelCode
FROM test_source;

接下来我们去查看yarn任务
2735D3C5-74DB-43DB-BB91-82DA077CACEB.png
点进去看看
开始向test写一些json数据
/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server 172.16.121.194:9092 --topic test
{"objId":"12345","data":"example data 1","capTime":"2023-11-07T08:00:00","dataType":"exampleType","channelCode":"ABCDEF"}
{"objId":"54321","data":"example data 2","capTime":"2023-11-07T08:15:00","dataType":"anotherType","channelCode":"GHIJKL"}
{"objId":"99999","data":"more example data","capTime":"2023-11-07T08:30:00","dataType":"additionalType","channelCode":"ZYXWVU"}
{"objId":"11111","data":"extra data","capTime":"2023-11-07T08:45:00","dataType":"extraType","channelCode":"QRSTUV"}
{"objId":"77777","data":"additional example data","capTime":"2023-11-07T09:00:00","dataType":"moreType","channelCode":"MNBVCX"}
{"objId":"88888","data":"more and more data","capTime":"2023-11-07T09:15:00","dataType":"typeX","channelCode":"POIUYT"}
{"objId":"22222","data":"different data","capTime":"2023-11-07T09:30:00","dataType":"typeY","channelCode":"LAKSDJ"}
{"objId":"66666","data":"sample data","capTime":"2023-11-07T09:45:00","dataType":"testType","channelCode":"QWERTY"}
{"objId":"44444","data":"new data","capTime":"2023-11-07T10:00:00","dataType":"newType","channelCode":"ZXCVBN"}
{"objId":"55555","data":"fresh data","capTime":"2023-11-07T10:15:00","dataType":"freshType","channelCode":"EDCRFV"}
7FBE0A68-D3AD-433C-8672-FB49C7C81FAA.png
查看flinkweb看数据过来了
2B6440A3-1BD4-4721-B9C4-3C4233BC02FF.png
输出到了控制台
AB8DC330-4763-4CE7-84D3-F09B86863507.png
完成


标签: 大数据运维

相关文章

EMR-java配置国密SM4加密

EMR-java配置国密SM4加密

首先找到bcprov-jdk15on-1.56.jar这个包<dependency>    <groupId>org.bouncycastle</groupId> ...

ES运维(七)添加sql插件

ES运维(七)添加sql插件

一、概述ElasticSearch安装SQL插件下载地址(中国大佬开发)二、集成sql插件1、下载es-sql插件下载地址: https://github.com/NLPchina/elastics...

ES运维(八)添加IK分词器

ES运维(八)添加IK分词器

一、概述ES自带standard analyzer、simple analyzer、whitespace analyzer、stop analyzer、language analyzer、patter...

运维监控体系总结

总结归纳运维工作中的监控内容。监控目标:明白监控的重要性以及使用监控要实现的业务目标通常包括以下三点:对目标系统进行实时监控监控可以实时反馈目标系统的当前状态 目标系统硬件、软件、业务是否正常、目前处...

nginx配置反向代理某个url

nginx配置反向代理某个url

本文讲的这个漏洞 主要是为了解决漏洞扫描的问题我先介绍下这个漏洞主要是因为访问https://172.16.120.17:18090/ws/v1/cluster/info这个 yarn rest的一个...

CDP实操--Ranger开启ldap认证

CDP实操--Ranger开启ldap认证

集群中已经部署了ldap主主模式,并且使用haproxy进行负载均衡,keepalive提供了虚拟ip。页面配置:Ranger进行同步用户:验证:使用ldap用户登录页面:FAQ:1、admin超级管...

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。