flinksql 读取kafka的key的字段，相关解析

清疚 2022-11-18 04:57 491阅读 0赞

1、因官网读取kafka的key结构文档说明比较复杂，研究半天才搞懂，特此进行记录

-- 创建数据源表
    CREATE TEMPORARY TABLE `source_kafka` (
      `time` varchar,
      `distinct_id` varchar,
     `msgkey` varchar
    ) with (
      'connector' = 'kafka',
       'topic' = 'hulin_sink_test',
      'properties.bootstrap.servers' = 'xxx:9092',
      'properties.group.id' = 'kafka_sink_test_by_wpp',
      -- 'format' = 'json',
      'key.format' = 'raw',
      'key.fields' = 'msgkey',
    
      'value.format' = 'json',
      --- 解析key的值是，要加上 'value.fields-include' = 'EXCEPT_KEY' 参数，不然这个 msgkey列也会被当成 value 的一部分参与 value 的解析，从而导致解析不出来数据
      'value.fields-include' = 'EXCEPT_KEY',
      'scan.startup.mode' = 'timestamp',
      'scan.startup.timestamp-millis' = '1618223534000'
    );
    
    
    -- 转化
    create TEMPORARY view source_kafka_change as 
    select
      cast(`time` as bigint) as `time`,
      `distinct_id`,
      cast(`msgkey` as varchar)  messageKey
    from `source_kafka`;
    
    
    -- 创建结果表
    CREATE TEMPORARY TABLE `sink_odps` (
      `userid` varchar,
      `time` bigint,
      messageKey varchar
    ) with (
        'connector' = 'print'
    );
    
    -- 数据落地
    BEGIN STATEMENT SET;
    
    insert into sink_odps 
    select
      `distinct_id` as userid,
      `time`,
      messageKey
    from source_kafka_change;
    END;

发表评论取消回复

表情：

评论列表（有 0 条评论，491人围观）

还没有评论，来说两句吧...

相关阅读

相关 FlinkSQL读取Kafka数据写入Mysql

![1007094-20201209162713850-822342885.png][] ![b8aeb3266edc1616699cea035b818c31.png][]

╰+哭是因爲堅強的太久メ/ 2022年12月27日 09:15/ 0 赞/ 202 阅读

相关 flinksql 读取kafka的key的字段，相关解析

1、因官网读取kafka的key结构文档说明比较复杂，研究半天才搞懂，特此进行记录 -- 创建数据源表 CREATE TEMPORARY TABLE `sou

清疚/ 2022年11月18日 04:57/ 0 赞/ 492 阅读

相关 95-910-335-源码-FlinkSQL-Calcite-FlinkSQL解析

文章目录 1.美图 2.config.mpp 3.Parser.tdd 4.packge,class,imports 5.keywords

淡淡的烟草味﹌/ 2022年09月10日 11:11/ 0 赞/ 159 阅读

相关 Kafka深度解析

原创文章，转载请务必将下面这段话置于文章开头处。　　本文转发自[Jason’s Blog][Jason_s Blog]，[原文链接][Link 1]　[http://www

心已赠人/ 2022年07月17日 00:16/ 0 赞/ 275 阅读

相关 Kafka深度解析

[原文链接][Link 1]　[http://www.jasongj.com/2015/01/02/Kafka深度解析][Link 1] 背景介绍 Kafka简介

￡神魔★判官ぃ/ 2022年06月02日 03:17/ 0 赞/ 288 阅读

相关 Kafka深度解析

Kafka技术内幕笔记：分区模型 Kafka集群向多个消息代理服务器（brokerserver）组成，发布至Kafka集群的每条消息都有一个类别，用主题（topic

àì夳堔傛蜴生んèń/ 2022年04月25日 01:52/ 0 赞/ 307 阅读

相关 Kafka深度解析

本文转发自[技术世界][Link 1]，[原文链接][Link 2]　[http://www.jasongj.com/2015/01/02/Kafka深度解析][Link 2]

心已赠人/ 2021年11月22日 13:40/ 0 赞/ 413 阅读

相关 FlinkSQL Kafka to Hive

基于CDH平台，Hive2.1.1-CDH6.2.0 ，写入Hive简单操作。 [Hive Streaming][] import java.time.Dur

╰半夏微凉°/ 2021年09月24日 21:12/ 0 赞/ 478 阅读

相关 FlinkSQL Kafka to Kafka

[Apache Kafka SQL Connector][] FlinkSql 读 Kafka import org.apache.flink.streami

朴灿烈づ我的快乐病毒、/ 2021年09月24日 14:30/ 0 赞/ 420 阅读

相关 kafka 文件解析

1、topic 按照partition存储，每个partition为一个目录，partiton命名规则为topic名称+有序序号 ![这里写图片描述][70] 2、par

待我称王封你为后i/ 2021年09月22日 00:06/ 0 赞/ 464 阅读