IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
图片批量下载器
↓批量下载图片,美女图库↓
图片自动播放器
↓图片自动播放器↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> 大数据 -> flink获取kafka的key value timestamp header -> 正文阅读

[大数据]flink获取kafka的key value timestamp header

flilnk在消费kafka数据的时候,我们习惯性的add一个kafkaConsumer,似乎万年不变,但是单我们需求变化的时候,我们该怎么办?

DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>(
        "clicks",
        new SimpleStringSchema(),
        properties
));
stream.print("Kafka");

注意 这里的DataStreamSource 的类型是string,而这个string就是kafka的value

?现在有需求,我们要获取kafka的的timestamp,然后跟进入flink的时间的对比,看flink处理数据的延迟时间有多少?

那么我们如何获取这个timstamp呢?

?最简单的办法,百度。 这位是用scala写的。

flink读取kafka中的数据的所有信息_第一片心意的博客-CSDN博客_flink读取kafka的数据

我就用java写了个。


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;

import java.text.SimpleDateFormat;
import java.util.*;

public class SourceKafkaConsumerRecordTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "xxxxxx1:9092");
        properties.setProperty("group.id", "consumer-group");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "ia-label",
                new SimpleStringSchema(),
                properties
        );
        FlinkKafkaConsumer<MyConsumerRecord> consumer2 = new FlinkKafkaConsumer<>(
                "ia-label",
                new KafkaDeserializationSchema<MyConsumerRecord>() {
                    @Override
                    public boolean isEndOfStream(MyConsumerRecord s) {
                        return false;
                    }
                    @Override
                    public MyConsumerRecord deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
                        Headers headers = consumerRecord.headers();
                        HashMap<String, String> headerMap = new HashMap<>();
                        for (Header header : headers) {
                            headerMap.put(header.key(),new String(header.value()));
                        }
                        byte[] key1 = consumerRecord.key();
                        byte[] value1 = consumerRecord.value();
                        String key = key1==null?null:new String(key1);
                        String value = new String(value1);
                        String timeStamp=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(consumerRecord.timestamp());
                        MyConsumerRecord myConsumerRecord = new MyConsumerRecord(key, value, timeStamp, headerMap);
                        System.out.println(myConsumerRecord);
                        return myConsumerRecord;
                    }

                    @Override
                    public TypeInformation<MyConsumerRecord> getProducedType() {
                        return TypeInformation.of(MyConsumerRecord.class);
                    }
                },
                properties
        );
        consumer2.setStartFromEarliest();
        consumer.setStartFromEarliest();
//        DataStreamSource<String> stream = env.addSource(consumer);\
//        stream.print("Kafka");
        DataStreamSource<MyConsumerRecord> stream2 = env.addSource(consumer2);
        stream2.print("All");
        env.execute();
    }

    static class MyConsumerRecord{
        String key;
        String value;
        String timeStamp;
        Map<String,String> header;

        public MyConsumerRecord(String key, String value, String timeStamp, Map<String, String> header) {
            this.key = key;
            this.value = value;
            this.timeStamp = timeStamp;
            this.header = header;
        }

        @Override
        public String toString() {
            return "MyConsumerRecord{" +
                    "key='" + key + '\'' +
                    ", value='" + value + '\'' +
                    ", timeStamp='" + timeStamp + '\'' +
                    ", header=" + header +
                    '}';
        }
    }
}

打印输出如下

?ok。 还可以获取 partition offset 这些信息,看自己了。

  大数据 最新文章
实现Kafka至少消费一次
亚马逊云科技:还在苦于ETL?Zero ETL的时代
初探MapReduce
【SpringBoot框架篇】32.基于注解+redis实现
Elasticsearch:如何减少 Elasticsearch 集
Go redis操作
Redis面试题
专题五 Redis高并发场景
基于GBase8s和Calcite的多数据源查询
Redis——底层数据结构原理
上一篇文章      下一篇文章      查看所有文章
加:2022-09-24 21:03:44  更:2022-09-24 21:04:21 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2024年11日历 -2024/11/23 10:14:20-

图片自动播放器
↓图片自动播放器↓
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓
一键清除垃圾
↓轻轻一点,清除系统垃圾↓
图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码