【Flink】Flink SQL 自定义 Source format

转x33g5p2x  于2022-01-05 转载在 Flink  
字(7.8k)|赞(0)|评价(0)|浏览(692)

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

  1. import org.apache.flink.api.common.serialization.DeserializationSchema;
  2. import org.apache.flink.api.common.serialization.SerializationSchema;
  3. import org.apache.flink.api.common.typeinfo.TypeInformation;
  4. import org.apache.flink.configuration.ConfigOption;
  5. import org.apache.flink.configuration.ReadableConfig;
  6. import org.apache.flink.formats.json.JsonOptions;
  7. import org.apache.flink.formats.json.TimestampFormat;
  8. import org.apache.flink.table.api.ValidationException;
  9. import org.apache.flink.table.connector.ChangelogMode;
  10. import org.apache.flink.table.connector.format.DecodingFormat;
  11. import org.apache.flink.table.connector.source.DynamicTableSource;
  12. import org.apache.flink.table.data.RowData;
  13. import org.apache.flink.table.factories.DeserializationFormatFactory;
  14. import org.apache.flink.table.factories.DynamicTableFactory;
  15. import org.apache.flink.table.factories.FactoryUtil;
  16. import org.apache.flink.table.types.DataType;
  17. import org.apache.flink.table.types.logical.RowType;
  19. import java.util.Collections;
  20. import java.util.HashSet;
  21. import java.util.Set;
  23. import static org.apache.flink.formats.json.JsonOptions.*;
  25. /** * Table format factory for providing configured instances of JSON to RowData * {@link SerializationSchema} and {@link DeserializationSchema}. */
  26. public class NullFormatFactory implements
  27. 		DeserializationFormatFactory {
  28. 	// Factory 的唯一标识
  29. 	public static final String IDENTIFIER = "null";
  30. 	
  31. 	@SuppressWarnings("unchecked")
  32. 	@Override
  33. 	// 解码的入口方法 基本上属于固定写法
  34. 	public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
  35. 			DynamicTableFactory.Context context,
  36. 			ReadableConfig formatOptions) {
  37. 		FactoryUtil.validateFactoryOptions(this, formatOptions);
  38. 		validateFormatOptions(formatOptions);
  39. 		
  40. 		final boolean failOnMissingField = formatOptions.get(FAIL_ON_MISSING_FIELD);
  41. 		final boolean ignoreParseErrors = formatOptions.get(IGNORE_PARSE_ERRORS);
  42. 		TimestampFormat timestampOption = JsonOptions.getTimestampFormat(formatOptions);
  43. 		
  44. 		return new DecodingFormat<DeserializationSchema<RowData>>() {
  45. 			@Override
  46. 			public DeserializationSchema<RowData> createRuntimeDecoder(
  47. 					DynamicTableSource.Context context,//ScanRuntimeProviderContext
  48. 					DataType producedDataType) { // 表的字段名和数据类型
  49. 				final RowType rowType = (RowType) producedDataType.getLogicalType();
  50. 				final TypeInformation<RowData> rowDataTypeInfo =
  51. 						(TypeInformation<RowData>) context.createTypeInformation(producedDataType);
  52. 				return new NullRowDataDeserializationSchema(
  53. 						rowType,
  54. 						rowDataTypeInfo,
  55. 						failOnMissingField,
  56. 						ignoreParseErrors,
  57. 						timestampOption
  58. 				);
  59. 			}
  60. 			
  61. 			@Override
  62. 			public ChangelogMode getChangelogMode() {
  63. 				return ChangelogMode.insertOnly();
  64. 			}
  65. 		};
  66. 	}
  67. 	
  68. 	
  69. 	@Override
  70. 	public String factoryIdentifier() {
  71. 		return IDENTIFIER;
  72. 	}
  73. 	
  74. 	@Override
  75. 	public Set<ConfigOption<?>> requiredOptions() {
  76. 		return Collections.emptySet();
  77. 	}
  78. 	
  79. 	@Override
  80. 	public Set<ConfigOption<?>> optionalOptions() {
  81. 		Set<ConfigOption<?>> options = new HashSet<>();
  82. 		options.add(FAIL_ON_MISSING_FIELD);
  83. 		options.add(IGNORE_PARSE_ERRORS);
  84. 		options.add(TIMESTAMP_FORMAT);
  85. 		return options;
  86. 	}
  87. 	
  88. 	// ------------------------------------------------------------------------
  89. 	// Validation
  90. 	// ------------------------------------------------------------------------
  91. 	
  92. 	static void validateFormatOptions(ReadableConfig tableOptions) {
  93. 		boolean failOnMissingField = tableOptions.get(FAIL_ON_MISSING_FIELD);
  94. 		boolean ignoreParseErrors = tableOptions.get(IGNORE_PARSE_ERRORS);
  95. 		String timestampFormat = tableOptions.get(TIMESTAMP_FORMAT);
  96. 		if (ignoreParseErrors && failOnMissingField) {
  97. 			throw new ValidationException(FAIL_ON_MISSING_FIELD.key()
  98. 					+ " and "
  99. 					+ IGNORE_PARSE_ERRORS.key()
  100. 					+ " shouldn't both be true.");
  101. 		}
  102. 		if (!TIMESTAMP_FORMAT_ENUM.contains(timestampFormat)) {
  103. 			throw new ValidationException(String.format("Unsupported value '%s' for %s. Supported values are [SQL, ISO-8601].",
  104. 					timestampFormat, TIMESTAMP_FORMAT.key()));
  105. 		}
  106. 	}
  107. }

自定义 DeserializationSchema

  1. import org.apache.flink.annotation.Internal;
  2. import org.apache.flink.api.common.serialization.DeserializationSchema;
  3. import org.apache.flink.api.common.typeinfo.TypeInformation;
  4. import org.apache.flink.formats.json.TimestampFormat;
  5. import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
  6. import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
  7. import org.apache.flink.table.data.RowData;
  8. import org.apache.flink.table.types.logical.RowType;
  10. import java.io.IOException;
  11. import java.util.Objects;
  13. import static org.apache.flink.util.Preconditions.checkNotNull;
  15. @Internal
  16. public class NullRowDataDeserializationSchema implements DeserializationSchema<RowData> {
  17. 	private static final long serialVersionUID = 1L;
  18. 	
  19. 	/** * Flag indicating whether to fail if a field is missing. */
  20. 	private final boolean failOnMissingField;
  21. 	
  22. 	/** * Flag indicating whether to ignore invalid fields/rows (default: throw an exception). */
  23. 	private final boolean ignoreParseErrors;
  24. 	
  25. 	/** * TypeInformation of the produced {@link RowData}. **/
  26. 	private final TypeInformation<RowData> resultTypeInfo;
  27. 	
  28. 	/** * Runtime converter that converts {@link JsonNode}s into * objects of Flink SQL internal data structures. **/
  29. 	
  30. 	/** * Object mapper for parsing the JSON. */
  31. 	private final ObjectMapper objectMapper = new ObjectMapper();
  32. 	
  33. 	/** * Timestamp format specification which is used to parse timestamp. */
  34. 	private final TimestampFormat timestampFormat;
  35. 	
  36. 	public NullRowDataDeserializationSchema(
  37. 			RowType rowType,
  38. 			TypeInformation<RowData> resultTypeInfo,
  39. 			boolean failOnMissingField,
  40. 			boolean ignoreParseErrors,
  41. 			TimestampFormat timestampFormat) {
  42. 		if (ignoreParseErrors && failOnMissingField) {
  43. 			throw new IllegalArgumentException(
  44. 					"JSON format doesn't support failOnMissingField and ignoreParseErrors are both enabled.");
  45. 		}
  46. 		this.resultTypeInfo = checkNotNull(resultTypeInfo);
  47. 		this.failOnMissingField = failOnMissingField;
  48. 		this.ignoreParseErrors = ignoreParseErrors;
  49. 		this.timestampFormat = timestampFormat;
  50. 	}
  51. 	
  52. 	@Override
  53. 	// 这里其实是真正的反序列化逻辑，比如说将 json 拍平 (多层嵌套转化为一层嵌套 )
  54. 	// 这里是重点，记得关注重点
  55. 	public RowData deserialize(byte[] message) throws IOException {
  56. 		return null;
  57. 	}
  58. 	
  59. 	@Override
  60. 	public boolean isEndOfStream(RowData nextElement) {
  61. 		return false;
  62. 	}
  63. 	
  64. 	@Override
  65. 	public TypeInformation<RowData> getProducedType() {
  66. 		return resultTypeInfo;
  67. 	}
  68. 	
  69. 	@Override
  70. 	public boolean equals(Object o) {
  71. 		if (this == o) {
  72. 			return true;
  73. 		}
  74. 		if (o == null || getClass() != o.getClass()) {
  75. 			return false;
  76. 		}
  77. 		NullRowDataDeserializationSchema that = (NullRowDataDeserializationSchema) o;
  78. 		return failOnMissingField == that.failOnMissingField &&
  79. 				ignoreParseErrors == that.ignoreParseErrors &&
  80. 				resultTypeInfo.equals(that.resultTypeInfo) &&
  81. 				timestampFormat.equals(that.timestampFormat);
  82. 	}
  83. 	
  84. 	@Override
  85. 	public int hashCode() {
  86. 		return Objects.hash(failOnMissingField, ignoreParseErrors, resultTypeInfo, timestampFormat);
  87. 	}
  88. }

4.使用自定义 Format

  1. public class SqlKafka {
  2. 	public static void main(String[] args) throws Exception {
  3. 		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4. 		
  5. 		EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  6. 		StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, environmentSettings);
  7. 		// enable checkpointing
  8. 		Configuration configuration = tableEnv.getConfig().getConfiguration();
  9. 		configuration.set(
  10. 				ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE);
  11. 		configuration.set(
  12. 				ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(10));
  13. 		
  14. 		String sql = "CREATE TABLE sourcedata (`id` bigint,`status` int,`city_id` bigint,`courier_id` bigint,info_index int,order_id bigint,tableName String" +
  15. 				") WITH (" +
  16. 				"'connector' = 'kafka','topic' = 'canal_monitor_order'," +
  17. 				"'properties.bootstrap.servers' = 'bigdata-dev-mq:9092','properties.group.id' = 'testGroup'," +
  18. 				"'format' = 'null','scan.startup.mode' = 'earliest-offset')";
  19. 		tableEnv.executeSql(sql);
  20. 		......

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

然后就可以直接执行了

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/qq_21383435/article/details/122303419

内容来源于网络，如有侵权，请联系作者删除！

sqlflinkkafka

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com