Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[FLINK-18823][format] Support serialization for debezium-json format #13333

Merged
merged 3 commits into from Sep 10, 2020
Merged

[FLINK-18823][format] Support serialization for debezium-json format #13333

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f5080f2
[FLINK-18823][format] Support serialization for debezium-json format
SteNicholas Sep 6, 2020
6f902c6
[FLINK-18823][format] Support serialization for debezium-json format
SteNicholas Sep 9, 2020
bb34b6b
[FLINK-18823][format] Support serialization for debezium-json format
SteNicholas Sep 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
8 changes: 2 additions & 6 deletions ...-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonFormatFactory.java
View file
Open in desktop
Expand Up @@ -121,13 +121,9 @@ public ChangelogMode getChangelogMode() {
}

@Override
public SerializationSchema<RowData> createRuntimeEncoder(
DynamicTableSink.Context context, DataType consumedDataType) {
public SerializationSchema<RowData> createRuntimeEncoder(DynamicTableSink.Context context, DataType consumedDataType) {
final RowType rowType = (RowType) consumedDataType.getLogicalType();
return new DebeziumJsonSerializationSchema(
rowType,
timestampFormat
);
return new DebeziumJsonSerializationSchema(rowType, timestampFormat);
}
};
}
Expand Down
17 changes: 12 additions & 5 deletions ...src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerializationSchema.java
View file
Open in desktop
Expand Up @@ -42,8 +42,8 @@
public class DebeziumJsonSerializationSchema implements SerializationSchema<RowData> {
private static final long serialVersionUID = 1L;

private static final StringData OP_INSERT = StringData.fromString("INSERT"); // insert
private static final StringData OP_DELETE = StringData.fromString("DELETE"); // delete
private static final StringData OP_INSERT = StringData.fromString("c"); // insert
private static final StringData OP_DELETE = StringData.fromString("d"); // delete

/** The serializer to serialize Debezium JSON data. **/
private final JsonRowDataSerializationSchema jsonSerializer;
Expand All @@ -58,14 +58,20 @@ public DebeziumJsonSerializationSchema(RowType rowType, TimestampFormat timestam

@Override
public void open(InitializationContext context) {
genericRowData = new GenericRowData(2);
genericRowData = new GenericRowData(3);
}

@Override
public byte[] serialize(RowData rowData) {
try {
genericRowData.setField(0, rowData);
genericRowData.setField(1, rowKind2String(rowData.getRowKind()));
if (RowKind.INSERT == rowData.getRowKind()) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should also consider UPDATE_AFTER and UPDATE_BEFORE .

genericRowData.setField(0, null);
genericRowData.setField(1, rowData);
} else if (RowKind.DELETE == rowData.getRowKind()) {
genericRowData.setField(0, rowData);
genericRowData.setField(1, null);
}
genericRowData.setField(2, rowKind2String(rowData.getRowKind()));
return jsonSerializer.serialize(genericRowData);
} catch (Throwable t) {
throw new RuntimeException(format("Could not serialize row '%s'.", rowData), t);
Expand Down Expand Up @@ -106,6 +112,7 @@ private static RowType createJsonRowType(DataType databaseSchema) {
// Debezium JSON contains some other information, e.g. "source", "ts_ms"
// but we don't need them.
return (RowType) DataTypes.ROW(
DataTypes.FIELD("before", databaseSchema),
DataTypes.FIELD("after", databaseSchema),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is wrong. Debezium serialize insert data in "after", but delete data in "before".

DataTypes.FIELD("op", DataTypes.STRING())).getLogicalType();
}
Expand Down
40 changes: 20 additions & 20 deletions ...son/src/test/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerDeSchemaTest.java
View file
Open in desktop
Expand Up @@ -186,26 +186,26 @@ private void testSerializationDeserialization(String resourceFile, boolean schem
}

expected = Arrays.asList(
"{\"after\":{\"id\":101,\"name\":\"scooter\",\"description\":\"Small 2-wheel scooter\",\"weight\":3.14},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":102,\"name\":\"car battery\",\"description\":\"12V car battery\",\"weight\":8.1},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":103,\"name\":\"12-pack drill bits\",\"description\":\"12-pack of drill bits with sizes ranging from #40 to #3\",\"weight\":0.8},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":104,\"name\":\"hammer\",\"description\":\"12oz carpenter's hammer\",\"weight\":0.75},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":105,\"name\":\"hammer\",\"description\":\"14oz carpenter's hammer\",\"weight\":0.875},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":106,\"name\":\"hammer\",\"description\":\"16oz carpenter's hammer\",\"weight\":1.0},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":107,\"name\":\"rocks\",\"description\":\"box of assorted rocks\",\"weight\":5.3},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":108,\"name\":\"jacket\",\"description\":\"water resistent black wind breaker\",\"weight\":0.1},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":109,\"name\":\"spare tire\",\"description\":\"24 inch spare tire\",\"weight\":22.2},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":106,\"name\":\"hammer\",\"description\":\"16oz carpenter's hammer\",\"weight\":1.0},\"op\":\"DELETE\"}",
"{\"after\":{\"id\":106,\"name\":\"hammer\",\"description\":\"18oz carpenter hammer\",\"weight\":1.0},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":107,\"name\":\"rocks\",\"description\":\"box of assorted rocks\",\"weight\":5.3},\"op\":\"DELETE\"}",
"{\"after\":{\"id\":107,\"name\":\"rocks\",\"description\":\"box of assorted rocks\",\"weight\":5.1},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":110,\"name\":\"jacket\",\"description\":\"water resistent white wind breaker\",\"weight\":0.2},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.18},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":110,\"name\":\"jacket\",\"description\":\"water resistent white wind breaker\",\"weight\":0.2},\"op\":\"DELETE\"}",
"{\"after\":{\"id\":110,\"name\":\"jacket\",\"description\":\"new water resistent white wind breaker\",\"weight\":0.5},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.18},\"op\":\"DELETE\"}",
"{\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.17},\"op\":\"INSERT\"}",
"{\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.17},\"op\":\"DELETE\"}"
"{\"before\":null,\"after\":{\"id\":101,\"name\":\"scooter\",\"description\":\"Small 2-wheel scooter\",\"weight\":3.14},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":102,\"name\":\"car battery\",\"description\":\"12V car battery\",\"weight\":8.1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":103,\"name\":\"12-pack drill bits\",\"description\":\"12-pack of drill bits with sizes ranging from #40 to #3\",\"weight\":0.8},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":104,\"name\":\"hammer\",\"description\":\"12oz carpenter's hammer\",\"weight\":0.75},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":105,\"name\":\"hammer\",\"description\":\"14oz carpenter's hammer\",\"weight\":0.875},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":106,\"name\":\"hammer\",\"description\":\"16oz carpenter's hammer\",\"weight\":1.0},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":107,\"name\":\"rocks\",\"description\":\"box of assorted rocks\",\"weight\":5.3},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":108,\"name\":\"jacket\",\"description\":\"water resistent black wind breaker\",\"weight\":0.1},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":109,\"name\":\"spare tire\",\"description\":\"24 inch spare tire\",\"weight\":22.2},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":109,\"name\":\"spare tire\",\"description\":\"24 inch spare tire\",\"weight\":22.2},\"op\":\"d\"}",
"{\"before\":null,\"after\":{\"id\":109,\"name\":\"spare tire\",\"description\":\"24 inch spare tire\",\"weight\":22.2},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":109,\"name\":\"spare tire\",\"description\":\"24 inch spare tire\",\"weight\":22.2},\"op\":\"d\"}",
"{\"before\":null,\"after\":{\"id\":109,\"name\":\"spare tire\",\"description\":\"24 inch spare tire\",\"weight\":22.2},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":110,\"name\":\"jacket\",\"description\":\"water resistent white wind breaker\",\"weight\":0.2},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.18},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.18},\"op\":\"d\"}",
"{\"before\":null,\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.18},\"op\":\"c\"}",
"{\"before\":null,\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.18},\"op\":\"d\"}",
"{\"before\":null,\"after\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.18},\"op\":\"c\"}",
"{\"before\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.17},\"after\":null,\"op\":\"d\"}"
);
assertEquals(expected, actual);
}
Expand Down