1.Flink数据类型

概述

  在Flink内部处理数据时,涉及到数据的网络传输、数据的序列化及反序列化,Flink需要知道操作的数据类型,为了能够在分布式计算过程中对数据的类型进行管理和判断,Flink中定义了TypeInformation来对数据类型进行描述,通过TypeInfomation能够在数据处理之前将数据类型推断出来,而不是真正在触发计算后才识别出,这样可以有效避免用户在编写Flink应用的过程出现数据类型问题。
常用的 TypeInformation 有BasicTypeInfo、TupleTypeInfo、CaseClassTypeInfo、PojoTypeInfo
类等,针对这些常用TypeInfomation介绍如下:

  • Flink通过实现BasicTypeInfo数据类型,能够支持任意Java原生基本(或装箱)类型和String类型,例如:Integer,String,Double等,除了BasicTypeInfo外,类似的还BasicArrayTypeInfo,支持Java中数组和集合类型;
  • 通过定义TupleTypeInfo来支持Tuple类型的数据;
  • 通过CaseClassTypeInfo支持Scala Case Class ;
  • PojoTypeInfo可以识别任意的POJOs类,包括Java和Scala类,POJOs可以完成复杂数据架构的定义,但是在Flink中使用POJOs数据类型需要满足以下要求:
    • POJOs类必须是Public修饰且独立定义,不能是内部类;
    • POJOs 类中必须含有默认空构造器;
    • POJOs类中所有的Fields必须是Public或者具有Public修饰的getter和Setter方法;

注意

  在使用Java API开发Flink应用时,通常情况下Flink都能正常进行数据类型推断进而选择合适的serializers以及comparators,但是在定义函数时如果使用到了泛型(lambda),JVM就会出现类型擦除的问题,Flink就获取不到对应的类型信息,这就需要借助类型提示(Type Hints)来告诉系统函数中传入的参数类型信息和输出类型,进而对数据类型进行推断处理。

实践

非lambda写法

非lambda写法,Flink是能正常识别的

public class WcTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // source:hello flink
        DataStreamSource<String> ds = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> tuple2Ds = ds.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                for (String word : s.split(" ")) {
                    collector.collect(Tuple2.of(word, 1));
                }
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = tuple2Ds.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return stringIntegerTuple2.f0;
            }
        });

        keyedStream.sum(1).print();

        env.execute();

    }

}
2024-04-25 11:26:16,255 [flink-akka.actor.default-dispatcher-8] [org.apache.flink.runtime.executiongraph.ExecutionGraph] [INFO] - Keyed Aggregation -> Sink: Print to Std. Out (6/8) (5fc86f952e35ab7079f08070c6694019_e70bbd798b564e0a50e10e343f1ac56b_5_0) switched from INITIALIZING to RUNNING.
3> (hello,1)
3> (hello,2)
3> (hello,3)
5> (world,1)
3> (hello,4)

在这里插入图片描述

在这里插入图片描述

lambda写法

public class WcTestL {

  public static void main(String[] args) throws Exception {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      // source:hello flink
      DataStreamSource<String> ds = env.socketTextStream("localhost", 9999);

      SingleOutputStreamOperator<Tuple2<String, Integer>> tuple2Ds =
              ds.flatMap((s, collector) -> {
                  for (String word : s.split(" ")) {
                      collector.collect(Tuple2.of(word, 1));
                  }
              });

      KeyedStream<Tuple2<String, Integer>, String> keyedStream = tuple2Ds.keyBy(stringIntegerTuple2 -> stringIntegerTuple2.f0);

      keyedStream.sum(1).print();

      env.execute();
}

问题如下,如注意章节所说
在这里插入图片描述

lambda改进

package com.fun.luxshare.test;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 单词统计
 * 数据类型
 */
public class WcTestL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // source:hello flink
        DataStreamSource<String> ds = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<Tuple2<String, Long>> tuple2Ds =
                ds.flatMap((String s, Collector<Tuple2<String, Long>> collector) -> {
                    for (String world : s.split(" ")) {
                        collector.collect(Tuple2.of(world, 1L));
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        KeyedStream<Tuple2<String, Long>, String> keyedStream = tuple2Ds.keyBy(stringTuple2 -> stringTuple2.f0);

        keyedStream.sum(1).print();
        env.execute();
    }
}

在这里插入图片描述

结束

Flink数据类型

相关推荐

  1. Flink 数据类型 & TypeInformation信息

    2024-04-26 12:12:02       54 阅读
  2. Flink SQL】Flink SQL 基础概念:数据类型

    2024-04-26 12:12:02       40 阅读
  3. 1-07基本数据类型

    2024-04-26 12:12:02       49 阅读
  4. 25、Flink 支持的数据类型及序列化详解

    2024-04-26 12:12:02       26 阅读
  5. Flink CDC 1.18.1 Oracle 数据同步到postgresql

    2024-04-26 12:12:02       43 阅读
  6. 数据类型)前端八股文修炼Day1

    2024-04-26 12:12:02       29 阅读

最近更新

  1. docker php8.1+nginx base 镜像 dockerfile 配置

    2024-04-26 12:12:02       98 阅读
  2. Could not load dynamic library ‘cudart64_100.dll‘

    2024-04-26 12:12:02       106 阅读
  3. 在Django里面运行非项目文件

    2024-04-26 12:12:02       87 阅读
  4. Python语言-面向对象

    2024-04-26 12:12:02       96 阅读

热门阅读

  1. 【Redis】深度学习与实践指南系列

    2024-04-26 12:12:02       39 阅读
  2. Playwright

    2024-04-26 12:12:02       36 阅读
  3. GESP一级 - 第二章 - 第3节 - 数据类型的转换

    2024-04-26 12:12:02       25 阅读
  4. 笔记:Python循环结构编程题(练习题)

    2024-04-26 12:12:02       36 阅读
  5. MVVM开发模式的理解

    2024-04-26 12:12:02       37 阅读
  6. 现代软件为什么要采用微服架构

    2024-04-26 12:12:02       30 阅读
  7. wow-string-list文件说明

    2024-04-26 12:12:02       41 阅读
  8. 设置Ollama在局域网中访问的方法(Ubuntu)

    2024-04-26 12:12:02       32 阅读
  9. CentOS即将停服,国产化系统替代参考

    2024-04-26 12:12:02       40 阅读
  10. Pytorch:神经网络过程代码详解

    2024-04-26 12:12:02       27 阅读
  11. 算法之重:探寻开发者不可忽视的基石力量

    2024-04-26 12:12:02       33 阅读