diff --git a/docs/content.zh/docs/connectors/table/formats/json.md b/docs/content.zh/docs/connectors/table/formats/json.md index 3869144949b451..2f239e4f49c173 100644 --- a/docs/content.zh/docs/connectors/table/formats/json.md +++ b/docs/content.zh/docs/connectors/table/formats/json.md @@ -247,6 +247,13 @@ Format 参数 特性 -------- +### 被忽略的解析错误指标 + +当 `json.ignore-parse-errors` 为 `true` 时，JSON 格式会在连接器提供的反序列化指标组下注册计数器 +`json.numMessagesWithParseErrors`。该指标统计至少有一个解析或字段转换错误被忽略的输入消息数， +包括仍能输出数据、但部分字段被置为 null 的消息。最外层 JSON Array 按一条消息计数。 +`debezium-json`、`canal-json` 等 CDC 格式不上报该指标。 + ### 允许 json array 直接展开成多行数据 通常，我们假设 JSON 的最外层数据是一个 JSON Object。所以一条 JSON 会转换成一行结果。 diff --git a/docs/content/docs/connectors/table/formats/json.md b/docs/content/docs/connectors/table/formats/json.md index 7ead1ab90d825d..9f18f8bc1beb86 100644 --- a/docs/content/docs/connectors/table/formats/json.md +++ b/docs/content/docs/connectors/table/formats/json.md @@ -260,6 +260,14 @@ The following table lists the type mapping from Flink type to JSON type. Features -------- +### Ignored parse error metric + +When `json.ignore-parse-errors` is `true`, the JSON format registers the counter +`json.numMessagesWithParseErrors` under the connector's deserialization metric group. +It counts input messages in which at least one parse or field conversion error was ignored, +including messages that still produce rows with null fields. A top-level JSON array counts +as one message. CDC formats such as `debezium-json` and `canal-json` do not report this metric. + ### Allow top-level JSON Arrays Usually, we assume the top-level of JSON string is a stringified JSON object. Then this stringified JSON object can be converted into one SQL row. diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java index 769e721a14de19..47981dc15d1f1e 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java @@ -22,6 +22,7 @@ import org.apache.flink.api.common.serialization.DeserializationSchema; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.formats.common.TimestampFormat; +import org.apache.flink.metrics.Counter; import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.RowType; @@ -80,6 +81,12 @@ public abstract class AbstractJsonDeserializationSchema implements Deserializati private transient List reusableCollectList; + private boolean parseErrorMetricsEnabled; + + private transient Counter numMessagesWithParseErrors; + + private transient boolean messageHasParseError; + public AbstractJsonDeserializationSchema( RowType rowType, TypeInformation resultTypeInfo, @@ -109,6 +116,31 @@ public void open(InitializationContext context) throws Exception { } reusableCollectList = new ArrayList<>(); collector = new ListCollector<>(reusableCollectList); + if (parseErrorMetricsEnabled && ignoreParseErrors) { + numMessagesWithParseErrors = + context.getMetricGroup().addGroup("json").counter("numMessagesWithParseErrors"); + LOG.info("Registered metric json.numMessagesWithParseErrors for ignored parse errors."); + } + } + + // CDC formats own their message boundaries and opt in separately. + void enableParseErrorMetrics() { + parseErrorMetricsEnabled = true; + } + + protected final Runnable getParseErrorListener() { + return this::markParseError; + } + + protected final void markParseError() { + messageHasParseError = true; + } + + protected final void finishMessage() { + if (messageHasParseError && numMessagesWithParseErrors != null) { + numMessagesWithParseErrors.inc(); + } + messageHasParseError = false; } /** @@ -153,13 +185,19 @@ public boolean equals(Object o) { AbstractJsonDeserializationSchema that = (AbstractJsonDeserializationSchema) o; return failOnMissingField == that.failOnMissingField && ignoreParseErrors == that.ignoreParseErrors + && parseErrorMetricsEnabled == that.parseErrorMetricsEnabled && resultTypeInfo.equals(that.resultTypeInfo) && timestampFormat.equals(that.timestampFormat); } @Override public int hashCode() { - return Objects.hash(failOnMissingField, ignoreParseErrors, resultTypeInfo, timestampFormat); + return Objects.hash( + failOnMissingField, + ignoreParseErrors, + parseErrorMetricsEnabled, + resultTypeInfo, + timestampFormat); } /** diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java index 9fb1787dcd56b3..55905f17f63f18 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java @@ -84,23 +84,29 @@ public DeserializationSchema createRuntimeDecoder( final RowType rowType = (RowType) producedDataType.getLogicalType(); final TypeInformation rowDataTypeInfo = context.createTypeInformation(producedDataType); + final AbstractJsonDeserializationSchema schema; if (jsonParserEnabled) { - return new JsonParserRowDataDeserializationSchema( - rowType, - rowDataTypeInfo, - failOnMissingField, - ignoreParseErrors, - timestampOption, - toProjectedNames( - (RowType) physicalDataType.getLogicalType(), projections)); + schema = + new JsonParserRowDataDeserializationSchema( + rowType, + rowDataTypeInfo, + failOnMissingField, + ignoreParseErrors, + timestampOption, + toProjectedNames( + (RowType) physicalDataType.getLogicalType(), + projections)); } else { - return new JsonRowDataDeserializationSchema( - rowType, - rowDataTypeInfo, - failOnMissingField, - ignoreParseErrors, - timestampOption); + schema = + new JsonRowDataDeserializationSchema( + rowType, + rowDataTypeInfo, + failOnMissingField, + ignoreParseErrors, + timestampOption); } + schema.enableParseErrorMetrics(); + return schema; } @Override diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java index 8a21962ad7dd59..89b9a4f46965d9 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserRowDataDeserializationSchema.java @@ -43,6 +43,8 @@ @Internal public class JsonParserRowDataDeserializationSchema extends AbstractJsonDeserializationSchema { + private final JsonParserToRowDataConverters converters; + /** * Runtime converter that converts {@link JsonParser}s into objects of Flink SQL internal data * structures. @@ -66,10 +68,16 @@ public JsonParserRowDataDeserializationSchema( TimestampFormat timestampFormat, @Nullable String[][] projectedFields) { super(rowType, resultTypeInfo, failOnMissingField, ignoreParseErrors, timestampFormat); - this.runtimeConverter = + this.converters = new JsonParserToRowDataConverters( - failOnMissingField, ignoreParseErrors, timestampFormat) - .createConverter(projectedFields, checkNotNull(rowType)); + failOnMissingField, ignoreParseErrors, timestampFormat); + this.runtimeConverter = converters.createConverter(projectedFields, checkNotNull(rowType)); + } + + @Override + public void open(InitializationContext context) throws Exception { + super.open(context); + converters.setParseErrorListener(getParseErrorListener()); } @Override @@ -100,7 +108,10 @@ public void deserialize(byte[] message, Collector out) throws IOExcepti throw new IOException( format("Failed to deserialize JSON '%s'.", new String(message)), t); } + markParseError(); logParseErrorIfDebugEnabled(message, t); + } finally { + finishMessage(); } } diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java index 097ed8c021daa8..c4db65fa6ab2f5 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonParserToRowDataConverters.java @@ -67,6 +67,7 @@ import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_FORMAT; import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_WITH_LOCAL_TIMEZONE_FORMAT; import static org.apache.flink.formats.common.TimeFormats.SQL_TIME_FORMAT; +import static org.apache.flink.util.Preconditions.checkNotNull; /** Tool class used to convert fields from {@link JsonParser} to {@link RowData}. */ @Internal @@ -83,6 +84,9 @@ public class JsonParserToRowDataConverters implements Serializable { /** Timestamp format specification which is used to parse timestamp. */ private final TimestampFormat timestampFormat; + /** Notified when a field conversion error is ignored and the field is set to null. */ + private transient Runnable parseErrorListener = () -> {}; + public JsonParserToRowDataConverters( boolean failOnMissingField, boolean ignoreParseErrors, @@ -92,6 +96,10 @@ public JsonParserToRowDataConverters( this.timestampFormat = timestampFormat; } + void setParseErrorListener(Runnable parseErrorListener) { + this.parseErrorListener = checkNotNull(parseErrorListener); + } + /** * Runtime converter that converts {@link JsonParser}s into objects of Flink Table & SQL * internal data structures. Unlike {@link JsonToRowDataConverters.JsonToRowDataConverter}, this @@ -473,6 +481,7 @@ private JsonParserToRowDataConverter wrapIntoNullableConverter( if (!ignoreParseErrors) { throw t; } + parseErrorListener.run(); return null; } }; diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java index a34d4f8d9b6871..d1fdedea501ca1 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDataDeserializationSchema.java @@ -46,6 +46,8 @@ public class JsonRowDataDeserializationSchema extends AbstractJsonDeserializationSchema { private static final long serialVersionUID = 1L; + private final JsonToRowDataConverters converters; + /** * Runtime converter that converts {@link JsonNode}s into objects of Flink SQL internal data * structures. @@ -59,9 +61,15 @@ public JsonRowDataDeserializationSchema( boolean ignoreParseErrors, TimestampFormat timestampFormat) { super(rowType, resultTypeInfo, failOnMissingField, ignoreParseErrors, timestampFormat); - this.runtimeConverter = - new JsonToRowDataConverters(failOnMissingField, ignoreParseErrors, timestampFormat) - .createConverter(checkNotNull(rowType)); + this.converters = + new JsonToRowDataConverters(failOnMissingField, ignoreParseErrors, timestampFormat); + this.runtimeConverter = converters.createConverter(checkNotNull(rowType)); + } + + @Override + public void open(InitializationContext context) throws Exception { + super.open(context); + converters.setParseErrorListener(getParseErrorListener()); } @Override @@ -84,6 +92,7 @@ public void deserialize(@Nullable byte[] message, Collector out) throws // will be caught by outer try-catch throw t; } + markParseError(); logParseErrorIfDebugEnabled(message, t); } } @@ -98,7 +107,10 @@ public void deserialize(@Nullable byte[] message, Collector out) throws throw new IOException( format("Failed to deserialize JSON '%s'.", new String(message)), t); } + markParseError(); logParseErrorIfDebugEnabled(message, t); + } finally { + finishMessage(); } } diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java index dedc58e0bff8e2..83f4c5f093a2a2 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonToRowDataConverters.java @@ -65,6 +65,7 @@ import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_FORMAT; import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_WITH_LOCAL_TIMEZONE_FORMAT; import static org.apache.flink.formats.common.TimeFormats.SQL_TIME_FORMAT; +import static org.apache.flink.util.Preconditions.checkNotNull; /** Tool class used to convert from {@link JsonNode} to {@link RowData}. * */ @Internal @@ -81,6 +82,9 @@ public class JsonToRowDataConverters implements Serializable { /** Timestamp format specification which is used to parse timestamp. */ private final TimestampFormat timestampFormat; + /** Notified when a field conversion error is ignored and the field is set to null. */ + private transient Runnable parseErrorListener = () -> {}; + public JsonToRowDataConverters( boolean failOnMissingField, boolean ignoreParseErrors, @@ -90,6 +94,10 @@ public JsonToRowDataConverters( this.timestampFormat = timestampFormat; } + void setParseErrorListener(Runnable parseErrorListener) { + this.parseErrorListener = checkNotNull(parseErrorListener); + } + /** * Runtime converter that converts {@link JsonNode}s into objects of Flink Table & SQL internal * data structures. @@ -394,6 +402,7 @@ private JsonToRowDataConverter wrapIntoNullableConverter(JsonToRowDataConverter if (!ignoreParseErrors) { throw t; } + parseErrorListener.run(); return null; } }; diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java index f118343609961c..5c42e796d9de6f 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonFormatFactoryTest.java @@ -165,6 +165,7 @@ private void testSchemaDeserializationSchema(Map options) { false, true, TimestampFormat.ISO_8601); + expectedDeser.enableParseErrorMetrics(); DeserializationSchema actualDeser = createTableSource(options) @@ -252,7 +253,7 @@ private void testJsonParserConfiguration(boolean enabled, Class expectedClass ScanRuntimeProviderContext.INSTANCE, SCHEMA.toPhysicalRowDataType()); - DeserializationSchema expectedDeser = + AbstractJsonDeserializationSchema expectedDeser = enabled ? new JsonParserRowDataDeserializationSchema( PHYSICAL_TYPE, @@ -266,6 +267,7 @@ private void testJsonParserConfiguration(boolean enabled, Class expectedClass false, true, TimestampFormat.ISO_8601); + expectedDeser.enableParseErrorMetrics(); assertThat(actualDeser).isInstanceOf(expectedClass); assertThat(actualDeser).isEqualTo(expectedDeser); diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonParseErrorMetricsTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonParseErrorMetricsTest.java new file mode 100644 index 00000000000000..d5c578d3e223a2 --- /dev/null +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonParseErrorMetricsTest.java @@ -0,0 +1,127 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.flink.formats.json; + +import org.apache.flink.api.common.functions.util.ListCollector; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.testutils.formats.DummyInitializationContext; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.testutils.MetricListener; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext; +import org.apache.flink.table.types.DataType; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.flink.table.api.DataTypes.FIELD; +import static org.apache.flink.table.api.DataTypes.INT; +import static org.apache.flink.table.api.DataTypes.ROW; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests metrics for ignored JSON parsing and conversion errors. */ +class JsonParseErrorMetricsTest { + + private static final String METRIC_NAME = "numMessagesWithParseErrors"; + + private static final DataType ROW_TYPE = ROW(FIELD("id", INT()), FIELD("age", INT())); + + private final MetricListener metricListener = new MetricListener(); + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testCountsAffectedMessages(boolean jsonParserEnabled) throws Exception { + final DeserializationSchema schema = createAndOpenSchema(jsonParserEnabled, true); + final Counter counter = metricListener.getCounter("json", METRIC_NAME).orElseThrow(); + + assertThat(deserialize(schema, "{\"id\":1,\"age\":2}")).hasSize(1); + assertThat(counter.getCount()).isZero(); + final List invalidFields = deserialize(schema, "{\"id\":\"bad\",\"age\":\"bad\"}"); + assertThat(invalidFields).hasSize(1); + assertThat(invalidFields.get(0).isNullAt(0)).isTrue(); + assertThat(invalidFields.get(0).isNullAt(1)).isTrue(); + assertThat(counter.getCount()).isEqualTo(1); + + assertThat(deserialize(schema, "not valid json")).isEmpty(); + assertThat(counter.getCount()).isEqualTo(2); + assertThat(deserialize(schema, "{\"id\":3}").get(0).getInt(0)).isEqualTo(3); + assertThat(counter.getCount()).isEqualTo(2); + + deserialize(schema, "[{\"id\":\"bad\"},{\"age\":\"bad\"},{\"id\":4}]"); + assertThat(counter.getCount()).isEqualTo(3); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testAcceptedNullsAndEmptyInputs(boolean jsonParserEnabled) throws Exception { + final DeserializationSchema schema = createAndOpenSchema(jsonParserEnabled, true); + final Counter counter = metricListener.getCounter("json", METRIC_NAME).orElseThrow(); + final List output = new ArrayList<>(); + schema.deserialize(null, new ListCollector<>(output)); + for (String message : new String[] {"", "[]", "{}", "{\"id\":null}"}) { + deserialize(schema, message); + } + assertThat(output).isEmpty(); + assertThat(counter.getCount()).isZero(); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testStrictMode(boolean jsonParserEnabled) throws Exception { + final DeserializationSchema schema = createAndOpenSchema(jsonParserEnabled, false); + assertThat(metricListener.getCounter("json", METRIC_NAME)).isEmpty(); + assertThatThrownBy(() -> deserialize(schema, "{\"id\":\"bad\"}")) + .isInstanceOf(IOException.class); + } + + private DeserializationSchema createAndOpenSchema( + boolean jsonParserEnabled, boolean ignoreErrors) throws Exception { + final Configuration options = new Configuration(); + options.set(JsonFormatOptions.IGNORE_PARSE_ERRORS, ignoreErrors); + options.set(JsonFormatOptions.DECODE_JSON_PARSER_ENABLED, jsonParserEnabled); + final DeserializationSchema schema = + new JsonFormatFactory() + .createDecodingFormat(null, options) + .createRuntimeDecoder(ScanRuntimeProviderContext.INSTANCE, ROW_TYPE); + schema.open( + new DummyInitializationContext() { + @Override + public MetricGroup getMetricGroup() { + return metricListener.getMetricGroup(); + } + }); + return schema; + } + + private static List deserialize(DeserializationSchema schema, String message) + throws IOException { + final List output = new ArrayList<>(); + schema.deserialize(message.getBytes(UTF_8), new ListCollector<>(output)); + return output; + } +}