1.概述

转载：Flink SQL 自定义 Source format

1.背景

由于 kafka 中的 json 属于嵌套，又不想二次序列化再把它展开，故自定义 format。

2.步骤

1.自定义 Factory 实现 DeserializationFormatFactory

2.自定义 DeserializationSchema 实现 DeserializationSchema

自定义 Factory 中 createDecodingFormat 方法返回 createDecodingFormat

3.自定义 Format

为了简单起见，我们自定义一个 NullFormat ，也就是无论 kafka 中的消息是什么都返回 null，相当于 kafka 中没有消息

自定义 Factory

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.formats.json.JsonOptions;
import org.apache.flink.formats.json.TimestampFormat;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;import java.util.Collections;
import java.util.HashSet;
import java.util.Set;import static org.apache.flink.formats.json.JsonOptions.*;/*** Table format factory for providing configured instances of JSON to RowData* {@link SerializationSchema} and {@link DeserializationSchema}.*/
public class NullFormatFactory implementsDeserializationFormatFactory {// Factory 的唯一标识public static final String IDENTIFIER = "null";@SuppressWarnings("unchecked")@Override// 解码的入口方法 基本上属于固定写法public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(DynamicTableFactory.Context context,ReadableConfig formatOptions) {FactoryUtil.validateFactoryOptions(this, formatOptions);validateFormatOptions(formatOptions);final boolean failOnMissingField = formatOptions.get(FAIL_ON_MISSING_FIELD);final boolean ignoreParseErrors = formatOptions.get(IGNORE_PARSE_ERRORS);TimestampFormat timestampOption = JsonOptions.getTimestampFormat(formatOptions);return new DecodingFormat<DeserializationSchema<RowData>>() {@Overridepublic DeserializationSchema<RowData> createRuntimeDecoder(DynamicTableSource.Context context,//ScanRuntimeProviderContextDataType producedDataType) { // 表的字段名和数据类型final RowType rowType = (RowType) producedDataType.getLogicalType();final TypeInformation<RowData> rowDataTypeInfo =(TypeInformation<RowData>) context.createTypeInformation(producedDataType);return new NullRowDataDeserializationSchema(rowType,rowDataTypeInfo,failOnMissingField,ignoreParseErrors,timestampOption);}@Overridepublic ChangelogMode getChangelogMode() {return ChangelogMode.insertOnly();}};}@Overridepublic String factoryIdentifier() {return IDENTIFIER;}@Overridepublic Set<ConfigOption<?>> requiredOptions() {return Collections.emptySet();}@Overridepublic Set<ConfigOption<?>> optionalOptions() {Set<ConfigOption<?>> options = new HashSet<>();options.add(FAIL_ON_MISSING_FIELD);options.add(IGNORE_PARSE_ERRORS);options.add(TIMESTAMP_FORMAT);return options;}// ------------------------------------------------------------------------//  Validation// ------------------------------------------------------------------------static void validateFormatOptions(ReadableConfig tableOptions) {boolean failOnMissingField = tableOptions.get(FAIL_ON_MISSING_FIELD);boolean ignoreParseErrors = tableOptions.get(IGNORE_PARSE_ERRORS);String timestampFormat = tableOptions.get(TIMESTAMP_FORMAT);if (ignoreParseErrors && failOnMissingField) {throw new ValidationException(FAIL_ON_MISSING_FIELD.key()+ " and "+ IGNORE_PARSE_ERRORS.key()+ " shouldn't both be true.");}if (!TIMESTAMP_FORMAT_ENUM.contains(timestampFormat)) {throw new ValidationException(String.format("Unsupported value '%s' for %s. Supported values are [SQL, ISO-8601].",timestampFormat, TIMESTAMP_FORMAT.key()));}}
}

自定义 DeserializationSchema

import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.formats.json.TimestampFormat;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;import java.io.IOException;
import java.util.Objects;import static org.apache.flink.util.Preconditions.checkNotNull;@Internal
public class NullRowDataDeserializationSchema implements DeserializationSchema<RowData> {private static final long serialVersionUID = 1L;/*** Flag indicating whether to fail if a field is missing.*/private final boolean failOnMissingField;/*** Flag indicating whether to ignore invalid fields/rows (default: throw an exception).*/private final boolean ignoreParseErrors;/*** TypeInformation of the produced {@link RowData}.**/private final TypeInformation<RowData> resultTypeInfo;/*** Runtime converter that converts {@link JsonNode}s into* objects of Flink SQL internal data structures.**//*** Object mapper for parsing the JSON.*/private final ObjectMapper objectMapper = new ObjectMapper();/*** Timestamp format specification which is used to parse timestamp.*/private final TimestampFormat timestampFormat;public NullRowDataDeserializationSchema(RowType rowType,TypeInformation<RowData> resultTypeInfo,boolean failOnMissingField,boolean ignoreParseErrors,TimestampFormat timestampFormat) {if (ignoreParseErrors && failOnMissingField) {throw new IllegalArgumentException("JSON format doesn't support failOnMissingField and ignoreParseErrors are both enabled.");}this.resultTypeInfo = checkNotNull(resultTypeInfo);this.failOnMissingField = failOnMissingField;this.ignoreParseErrors = ignoreParseErrors;this.timestampFormat = timestampFormat;}@Override// 这里其实是真正的反序列化逻辑，比如说将 json 拍平 (多层嵌套转化为一层嵌套 )// 这里是重点，记得关注重点public RowData deserialize(byte[] message) throws IOException {return null;}@Overridepublic boolean isEndOfStream(RowData nextElement) {return false;}@Overridepublic TypeInformation<RowData> getProducedType() {return resultTypeInfo;}@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (o == null || getClass() != o.getClass()) {return false;}NullRowDataDeserializationSchema that = (NullRowDataDeserializationSchema) o;return failOnMissingField == that.failOnMissingField &&ignoreParseErrors == that.ignoreParseErrors &&resultTypeInfo.equals(that.resultTypeInfo) &&timestampFormat.equals(that.timestampFormat);}@Overridepublic int hashCode() {return Objects.hash(failOnMissingField, ignoreParseErrors, resultTypeInfo, timestampFormat);}
}

4.使用自定义 Format

public class SqlKafka {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, environmentSettings);// enable checkpointingConfiguration configuration = tableEnv.getConfig().getConfiguration();configuration.set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE);configuration.set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10));String sql = "CREATE TABLE sourcedata (`id` bigint,`status` int,`city_id` bigint,`courier_id` bigint,info_index int,order_id bigint,tableName String" +") WITH (" +"'connector' = 'kafka','topic' = 'canal_monitor_order'," +"'properties.bootstrap.servers' = 'bigdata-dev-mq:9092','properties.group.id' = 'testGroup'," +"'format' = 'null','scan.startup.mode' = 'earliest-offset')";tableEnv.executeSql(sql);......

‘format’ = ‘null’ Factory 的唯一标识

然后就可以直接执行了

【Flink】Flink SQL 自定义 Source format相关推荐

Flink SQL 自定义 Connector
文章目录一.Flink Table 实现架构图: 二.Flink Table Conncetor流程解析 1.定义动态工厂类 2.定义Connector Source/Sink类 (1)定义Conn ...
Flink SQL自定义connector
本文翻译自:Flink v1.11官方文档动态表是Flink的Table&SQL API的核心概念,用于以统一方式处理有界和无界数据. 因为动态表只是一个逻辑概念,所以Flink并不拥有数据 ...
Flink 源码 | 自定义 Format 消费 Maxwell CDC 数据
Flink 1.11 最重要的 Feature -- Hive Streaming 之前已经和大家分享过了,今天就和大家来聊一聊另一个特别重要的功能 -- CDC. CDC概述何为CDC?Chang ...
《从0到1学习Flink》—— 如何自定义 Data Source ？
前言在 <从0到1学习Flink>-- Data Source 介绍文章中,我给大家介绍了 Flink Data Source 以及简短的介绍了一下自定义 Data Source,这篇 ...
Flink自带的Source源算子以及自定义数据源Source
文章目录 Flink的DataStream API(基础篇) Source源算子从集合中读取数据从文件中读取数据从Scoket中读取数据从Kafka中读取数据自定义Source Flink的 ...
Flink SQL自定义聚合函数
<2021年最新版大数据面试题全面开启更新> 本篇幅介绍Flink Table/SQL中如何自定义一个聚合函数,介绍其基本用法.撤回定义以及与源码结合分析每个方法的调用位置. 基本使用 F ...
实时数仓入门训练营：实时计算 Flink 版 SQL 实践
简介:<实时数仓入门训练营>由阿里云研究员王峰.阿里云高级产品专家刘一鸣等实时计算 Flink 版和 Hologres 的多名技术/产品一线专家齐上阵,合力搭建此次训练营的课程体系,精心打 ...
【Flink】Flink Table SQL 用户自定义函数: UDF、UDAF、UDTF
本文总结Flink Table & SQL中的用户自定义函数: UDF.UDAF.UDTF. UDF: 自定义标量函数(User Defined Scalar Function).一行输入一行 ...
flink之SQL入门
SQL部分学习 Table API的特点Table API和SQL都是Apache Flink中高等级的分析API,SQL所具备的特点Table API也都具有,如下: 声明式 - 用户只关心做什么, ...

【Flink】Flink SQL 自定义 Source format

1.概述

4.使用自定义 Format

【Flink】Flink SQL 自定义 Source format相关推荐

最新文章

热门文章