This worked, thanks! Looking forward to the future releases :) On Mon, Aug 31, 2020 at 5:06 PM Marta Paes Moreira <ma...@ververica.com> wrote:
> Hey, Rex! > > This is likely due to the tombstone records that Debezium produces for > DELETE operations (i.e. a record with the same key as the deleted row and a > value of null). These are markers for Kafka to indicate that log > compaction can remove all records for the given key, and the initial > implementation of the debezium-format can't handle them. This issue is > already documented (and solved) in [1]. > > In the meantime, can you try adding "tombstones.on.delete":false" to the > configuration of your Debezium MySQL connector? Marta > [1] https://issues.apache.org/jira/browse/FLINK-18705 > > On Tue, Sep 1, 2020 at 1:36 AM Rex Fenley <r...@remind101.com> wrote: > >> Hi, getting so close but ran into another issue: >> >> Flink successfully reads changes from Debezium/Kafka and writes them to >> Elasticsearch, but there's a problem with deletions. When I DELETE a row >> from MySQL the deletion makes it successfully all the way to Elasticsearch >> which is great, but then the taskmanager suddenly dies with a null pointer >> exception. Inserts and Updates do not have the same problem. This seems >> very odd. Any help would be much appreciated. Thanks! >> >> flink-taskmanager_1 | 2020-08-31 23:30:33,684 WARN >> org.apache.flink.runtime.taskmanager.Task [] - Source: >> TableSourceScan(table=[[default_catalog, default_database, >> topic_addresses]], fields=[id, customer_id, street, city, state, zip, >> type]) -> Sink: Sink(table=[default_catalog.default_database.ESAddresses], >> fields=[id, customer_id, street, city, state, zip, type]) (1/2) >> (2b79917cb528f37fad7f636740d2fdd8) switched from RUNNING to FAILED. >> flink-taskmanager_1 | java.lang.NullPointerException: null >> flink-taskmanager_1 | at java.lang.String.<init>(String.java:566) >> ~[?:1.8.0_265] >> flink-taskmanager_1 | at >> org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema.deserialize(DebeziumJsonDeserializationSchema.java:136) >> ~[flink-json-1.11.1.jar:1.11.1] >> flink-taskmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.internals.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:56) >> ~[flink-jdbc-test-assembly-0.1-SNAPSHOT.jar:0.1-SNAPSHOT] >> flink-taskmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:181) >> ~[flink-jdbc-test-assembly-0.1-SNAPSHOT.jar:0.1-SNAPSHOT] >> flink-taskmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141) >> ~[flink-jdbc-test-assembly-0.1-SNAPSHOT.jar:0.1-SNAPSHOT] >> flink-taskmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755) >> ~[flink-jdbc-test-assembly-0.1-SNAPSHOT.jar:0.1-SNAPSHOT] >> flink-taskmanager_1 | at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) >> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >> flink-taskmanager_1 | at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) >> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >> flink-taskmanager_1 | at >> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201) >> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >> flink-taskmanager_1 | 2020-08-31 23:30:33,720 INFO >> org.apache.flink.runtime.taskmanager.Task [] - Freeing >> task resources for Source: TableSourceScan(table=[[default_catalog, >> default_database, topic_addresses]], fields=[id, customer_id, street, city, >> state, zip, type]) -> Sink: >> Sink(table=[default_catalog.default_database.ESAddresses], fields=[id, >> customer_id, street, city, state, zip, type]) (1/2) >> (2b79917cb528f37fad7f636740d2fdd8). >> flink-taskmanager_1 | 2020-08-31 23:30:33,728 INFO >> org.apache.flink.runtime.taskexecutor.TaskExecutor [] - >> Un-registering task and sending final execution state FAILED to JobManager >> for task Source: TableSourceScan(table=[[default_catalog, default_database, >> topic_addresses]], fields=[id, customer_id, street, city, state, zip, >> type]) -> Sink: Sink(table=[default_catalog.default_database.ESAddresses], >> fields=[id, customer_id, street, city, state, zip, type]) (1/2) >> 2b79917cb528f37fad7f636740d2fdd8. >> flink-jobmanager_1 | 2020-08-31 23:30:33,770 INFO >> org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: >> TableSourceScan(table=[[default_catalog, default_database, >> topic_addresses]], fields=[id, customer_id, street, city, state, zip, >> type]) -> Sink: Sink(table=[default_catalog.default_database.ESAddresses], >> fields=[id, customer_id, street, city, state, zip, type]) (1/2) >> (2b79917cb528f37fad7f636740d2fdd8) switched from RUNNING to FAILED on >> org.apache.flink.runtime.jobmaster.slotpool.SingleLogicalSlot@2e246b35. >> flink-jobmanager_1 | java.lang.NullPointerException: null >> flink-jobmanager_1 | at java.lang.String.<init>(String.java:566) >> ~[?:1.8.0_265] >> flink-jobmanager_1 | at >> org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema.deserialize(DebeziumJsonDeserializationSchema.java:136) >> ~[flink-json-1.11.1.jar:1.11.1] >> flink-jobmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.internals.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:56) >> ~[?:?] >> flink-jobmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:181) >> ~[?:?] >> flink-jobmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141) >> ~[?:?] >> flink-jobmanager_1 | at >> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755) >> ~[?:?] >> flink-jobmanager_1 | at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) >> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >> flink-jobmanager_1 | at >> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) >> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >> flink-jobmanager_1 | at >> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201) >> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >> >> On Mon, Aug 31, 2020 at 12:27 PM Rex Fenley <r...@remind101.com> wrote: >> >>> Ah, my bad, thanks for pointing that out Arvid! >>> >>> On Mon, Aug 31, 2020 at 12:00 PM Arvid Heise <ar...@ververica.com> >>> wrote: >>> >>>> Hi Rex, >>>> >>>> you still forgot >>>> >>>> 'debezium-json.schema-include' = true >>>> >>>> Please reread my mail. >>>> >>>> >>>> On Mon, Aug 31, 2020 at 7:55 PM Rex Fenley <r...@remind101.com> wrote: >>>> >>>>> Thanks for the input, though I've certainly included a schema as is >>>>> reflected earlier in this thread. Including here again >>>>> ... >>>>> tableEnv.executeSql(""" >>>>> CREATE TABLE topic_addresses ( >>>>> -- schema is totally the same to the MySQL "addresses" table >>>>> id INT, >>>>> customer_id INT, >>>>> street STRING, >>>>> city STRING, >>>>> state STRING, >>>>> zip STRING, >>>>> type STRING, >>>>> PRIMARY KEY (id) NOT ENFORCED >>>>> ) WITH ( >>>>> 'connector' = 'kafka', >>>>> 'topic' = 'dbserver1.inventory.addresses', >>>>> 'properties.bootstrap.servers' = 'flink-jdbc-test_kafka_1:9092', >>>>> 'properties.group.id' = 'testGroup', >>>>> 'format' = 'debezium-json' -- using debezium-json as the format >>>>> ) >>>>> """) >>>>> >>>>> val table = tableEnv.from("topic_addresses").select($"*") >>>>> ... >>>>> >>>>> On Mon, Aug 31, 2020 at 2:39 AM Arvid Heise <ar...@ververica.com> >>>>> wrote: >>>>> >>>>>> Hi Rex, >>>>>> >>>>>> the connector expects a value without a schema, but the message >>>>>> contains a schema. You can tell Flink that the schema is included as >>>>>> written in the documentation [1]. >>>>>> >>>>>> CREATE TABLE topic_products ( >>>>>> -- schema is totally the same to the MySQL "products" table >>>>>> id BIGINT, >>>>>> name STRING, >>>>>> description STRING, >>>>>> weight DECIMAL(10, 2)) WITH ( >>>>>> 'connector' = 'kafka', >>>>>> 'topic' = 'products_binlog', >>>>>> 'properties.bootstrap.servers' = 'localhost:9092', >>>>>> 'properties.group.id' = 'testGroup', >>>>>> 'format' = 'debezium-json', >>>>>> 'debezium-json.schema-include' = true) >>>>>> >>>>>> @Jark Wu <imj...@gmail.com> , it would be probably good to make the >>>>>> connector more robust and catch these types of misconfigurations. >>>>>> >>>>>> [1] >>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/debezium.html#how-to-use-debezium-format >>>>>> >>>>>> On Fri, Aug 28, 2020 at 11:56 PM Rex Fenley <r...@remind101.com> >>>>>> wrote: >>>>>> >>>>>>> Awesome, so that took me a step further. When running i'm receiving >>>>>>> an error however. FYI, my docker-compose file is based on the Debezium >>>>>>> mysql tutorial which can be found here >>>>>>> https://debezium.io/documentation/reference/1.2/tutorial.html >>>>>>> >>>>>>> Part of the stack trace: >>>>>>> >>>>>>> flink-jobmanager_1 | Caused by: java.io.IOException: Corrupt >>>>>>> Debezium JSON message >>>>>>> '{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"int32","optional":false,"field":"customer_id"},{"type":"string","optional":false,"field":"street"},{"type":"string","optional":false,"field":"city"},{"type":"string","optional":false,"field":"state"},{"type":"string","optional":false,"field":"zip"},{"type":"string","optional":false,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"SHIPPING,BILLING,LIVING"},"field":"type"}],"optional":true,"name":"dbserver1.inventory.addresses.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"int32","optional":false,"field":"customer_id"},{"type":"string","optional":false,"field":"street"},{"type":"string","optional":false,"field":"city"},{"type":"string","optional":false,"field":"state"},{"type":"string","optional":false,"field":"zip"},{"type":"string","optional":false,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"SHIPPING,BILLING,LIVING"},"field":"type"}],"optional":true,"name":"dbserver1.inventory.addresses.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.addresses.Envelope"},"payload":{"before":null,"after":{"id":18,"customer_id":1004,"street":"111 >>>>>>> cool street","city":"Big >>>>>>> City","state":"California","zip":"90000","type":"BILLING"},"source":{"version":"1.2.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1598651432000,"snapshot":"false","db":"inventory","table":"addresses","server_id":223344,"gtid":null,"file":"mysql-bin.000010","pos":369,"row":0,"thread":5,"query":null},"op":"c","ts_ms":1598651432407,"transaction":null}}'. >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema.deserialize(DebeziumJsonDeserializationSchema.java:136) >>>>>>> ~[flink-json-1.11.1.jar:1.11.1] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.internals.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:56) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:181) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) >>>>>>> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) >>>>>>> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201) >>>>>>> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >>>>>>> flink-jobmanager_1 | Caused by: java.lang.NullPointerException >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema.deserialize(DebeziumJsonDeserializationSchema.java:115) >>>>>>> ~[flink-json-1.11.1.jar:1.11.1] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.internals.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:56) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:181) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755) >>>>>>> ~[?:?] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) >>>>>>> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) >>>>>>> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >>>>>>> flink-jobmanager_1 | at >>>>>>> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201) >>>>>>> ~[flink-dist_2.12-1.11.1.jar:1.11.1] >>>>>>> >>>>>>> On Thu, Aug 27, 2020 at 8:12 PM Jark Wu <imj...@gmail.com> wrote: >>>>>>> >>>>>>>> Hi, >>>>>>>> >>>>>>>> This is a known issue in 1.11.0, and has been fixed in 1.11.1. >>>>>>>> >>>>>>>> >>>>>>>> Best, >>>>>>>> Jark >>>>>>>> >>>>>>>> On Fri, 28 Aug 2020 at 06:52, Rex Fenley <r...@remind101.com> wrote: >>>>>>>> >>>>>>>>> Hi again! >>>>>>>>> >>>>>>>>> I'm tested out locally in docker on Flink 1.11 first to get my >>>>>>>>> bearings before downgrading to 1.10 and figuring out how to replace >>>>>>>>> the >>>>>>>>> Debezium connector. However, I'm getting the following error >>>>>>>>> ``` >>>>>>>>> Provided trait [BEFORE_AND_AFTER] can't satisfy required trait >>>>>>>>> [ONLY_UPDATE_AFTER]. This is a bug in planner, please file an issue. >>>>>>>>> ``` >>>>>>>>> >>>>>>>>> Any suggestions for me to fix this? >>>>>>>>> >>>>>>>>> code: >>>>>>>>> >>>>>>>>> val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment >>>>>>>>> val blinkStreamSettings = >>>>>>>>> EnvironmentSettings >>>>>>>>> .newInstance() >>>>>>>>> .useBlinkPlanner() >>>>>>>>> .inStreamingMode() >>>>>>>>> .build() >>>>>>>>> val tableEnv = StreamTableEnvironment.create(bsEnv, >>>>>>>>> blinkStreamSettings) >>>>>>>>> >>>>>>>>> // Table from Debezium mysql example docker: >>>>>>>>> // >>>>>>>>> +-------------+-------------------------------------+------+-----+---------+----------------+ >>>>>>>>> // | Field | Type | Null | Key | Default | Extra | >>>>>>>>> // >>>>>>>>> +-------------+-------------------------------------+------+-----+---------+----------------+ >>>>>>>>> // | id | int(11) | NO | PRI | NULL | auto_increment | >>>>>>>>> // | customer_id | int(11) | NO | MUL | NULL | | >>>>>>>>> // | street | varchar(255) | NO | | NULL | | >>>>>>>>> // | city | varchar(255) | NO | | NULL | | >>>>>>>>> // | state | varchar(255) | NO | | NULL | | >>>>>>>>> // | zip | varchar(255) | NO | | NULL | | >>>>>>>>> // | type | enum('SHIPPING','BILLING','LIVING') | NO | | NULL | | >>>>>>>>> // >>>>>>>>> +-------------+-------------------------------------+------+-----+---------+----------------+ >>>>>>>>> >>>>>>>>> tableEnv.executeSql(""" >>>>>>>>> CREATE TABLE topic_addresses ( >>>>>>>>> -- schema is totally the same to the MySQL "addresses" table >>>>>>>>> id INT, >>>>>>>>> customer_id INT, >>>>>>>>> street STRING, >>>>>>>>> city STRING, >>>>>>>>> state STRING, >>>>>>>>> zip STRING, >>>>>>>>> type STRING, >>>>>>>>> PRIMARY KEY (id) NOT ENFORCED >>>>>>>>> ) WITH ( >>>>>>>>> 'connector' = 'kafka', >>>>>>>>> 'topic' = 'dbserver1.inventory.addresses', >>>>>>>>> 'properties.bootstrap.servers' = 'flink-jdbc-test_kafka_1:9092', >>>>>>>>> 'properties.group.id' = 'testGroup', >>>>>>>>> 'format' = 'debezium-json' -- using debezium-json as the format >>>>>>>>> ) >>>>>>>>> """) >>>>>>>>> >>>>>>>>> val table = tableEnv.from("topic_addresses").select($"*") >>>>>>>>> >>>>>>>>> // Defining a PK automatically puts it in Upsert mode, which we >>>>>>>>> want. >>>>>>>>> // TODO: type should be a keyword, is that acceptable by the DDL? >>>>>>>>> tableEnv.executeSql(""" >>>>>>>>> CREATE TABLE ESAddresses ( >>>>>>>>> id INT, >>>>>>>>> customer_id INT, >>>>>>>>> street STRING, >>>>>>>>> city STRING, >>>>>>>>> state STRING, >>>>>>>>> zip STRING, >>>>>>>>> type STRING, >>>>>>>>> PRIMARY KEY (id) NOT ENFORCED >>>>>>>>> ) WITH ( >>>>>>>>> 'connector' = 'elasticsearch-7', >>>>>>>>> 'hosts' = 'http://flink-jdbc-test_graph-elasticsearch_1:9200', >>>>>>>>> 'index' = 'flinkaddresses', >>>>>>>>> 'format' = 'json' >>>>>>>>> ) >>>>>>>>> """) >>>>>>>>> >>>>>>>>> table.executeInsert("ESAddresses").print() >>>>>>>>> >>>>>>>>> Thanks! >>>>>>>>> >>>>>>>>> On Thu, Aug 27, 2020 at 11:53 AM Rex Fenley <r...@remind101.com> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>>> Thanks! >>>>>>>>>> >>>>>>>>>> On Thu, Aug 27, 2020 at 5:33 AM Jark Wu <imj...@gmail.com> wrote: >>>>>>>>>> >>>>>>>>>>> Hi, >>>>>>>>>>> >>>>>>>>>>> Regarding the performance difference, the proposed way will have >>>>>>>>>>> one more stateful operator (deduplication) than the native 1.11 cdc >>>>>>>>>>> support. >>>>>>>>>>> The overhead of the deduplication operator is just similar to a >>>>>>>>>>> simple group by aggregate (max on each non-key column). >>>>>>>>>>> >>>>>>>>>>> Best, >>>>>>>>>>> Jark >>>>>>>>>>> >>>>>>>>>>> On Tue, 25 Aug 2020 at 02:21, Rex Fenley <r...@remind101.com> >>>>>>>>>>> wrote: >>>>>>>>>>> >>>>>>>>>>>> Thank you so much for the help! >>>>>>>>>>>> >>>>>>>>>>>> On Mon, Aug 24, 2020 at 4:08 AM Marta Paes Moreira < >>>>>>>>>>>> ma...@ververica.com> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Yes — you'll get the full row in the payload; and you can also >>>>>>>>>>>>> access the change operation, which might be useful in your case. >>>>>>>>>>>>> >>>>>>>>>>>>> About performance, I'm summoning Kurt and @Jark Wu >>>>>>>>>>>>> <j...@apache.org> to the thread, who will be able to give you >>>>>>>>>>>>> a more complete answer and likely also some optimization tips for >>>>>>>>>>>>> your >>>>>>>>>>>>> specific use case. >>>>>>>>>>>>> >>>>>>>>>>>>> Marta >>>>>>>>>>>>> >>>>>>>>>>>>> On Fri, Aug 21, 2020 at 8:55 PM Rex Fenley <r...@remind101.com> >>>>>>>>>>>>> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>>> Yup! This definitely helps and makes sense. >>>>>>>>>>>>>> >>>>>>>>>>>>>> The 'after' payload comes with all data from the row right? >>>>>>>>>>>>>> So essentially inserts and updates I can insert/replace data by >>>>>>>>>>>>>> pk and null >>>>>>>>>>>>>> values I just delete by pk, and then I can build out the rest of >>>>>>>>>>>>>> my joins >>>>>>>>>>>>>> like normal. >>>>>>>>>>>>>> >>>>>>>>>>>>>> Are there any performance implications of doing it this way >>>>>>>>>>>>>> that is different from the out-of-the-box 1.11 solution? >>>>>>>>>>>>>> >>>>>>>>>>>>>> On Fri, Aug 21, 2020 at 2:28 AM Marta Paes Moreira < >>>>>>>>>>>>>> ma...@ververica.com> wrote: >>>>>>>>>>>>>> >>>>>>>>>>>>>>> Hi, Rex. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Part of what enabled CDC support in Flink 1.11 was the >>>>>>>>>>>>>>> refactoring of the table source interfaces (FLIP-95 [1]), and >>>>>>>>>>>>>>> the new >>>>>>>>>>>>>>> ScanTableSource [2], which allows to emit bounded/unbounded >>>>>>>>>>>>>>> streams with >>>>>>>>>>>>>>> insert, update and delete rows. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> In theory, you could consume data generated with Debezium as >>>>>>>>>>>>>>> regular JSON-encoded events before Flink 1.11 — there just >>>>>>>>>>>>>>> wasn't a >>>>>>>>>>>>>>> convenient way to really treat it as "changelog". As a >>>>>>>>>>>>>>> workaround, what you >>>>>>>>>>>>>>> can do in Flink 1.10 is process these messages as JSON and >>>>>>>>>>>>>>> extract the >>>>>>>>>>>>>>> "after" field from the payload, and then apply de-duplication >>>>>>>>>>>>>>> [3] to keep >>>>>>>>>>>>>>> only the last row. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> The DDL for your source table would look something like: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> CREATE TABLE tablename ( *... * after ROW(`field1` >>>>>>>>>>>>>>> DATATYPE, `field2` DATATYPE, ...) ) WITH ( 'connector' = >>>>>>>>>>>>>>> 'kafka', 'format' = 'json', ... ); >>>>>>>>>>>>>>> Hope this helps! >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Marta >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> [1] >>>>>>>>>>>>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-95%3A+New+TableSource+and+TableSink+interfaces >>>>>>>>>>>>>>> [2] >>>>>>>>>>>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.11/api/java/org/apache/flink/table/connector/source/ScanTableSource.html >>>>>>>>>>>>>>> [3] >>>>>>>>>>>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/sql/queries.html#deduplication >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Fri, Aug 21, 2020 at 10:28 AM Chesnay Schepler < >>>>>>>>>>>>>>> ches...@apache.org> wrote: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> @Jark Would it be possible to use the 1.11 debezium support >>>>>>>>>>>>>>>> in 1.10? >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> On 20/08/2020 19:59, Rex Fenley wrote: >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Hi, >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> I'm trying to set up Flink with Debezium CDC Connector on >>>>>>>>>>>>>>>> AWS EMR, however, EMR only supports Flink 1.10.0, whereas >>>>>>>>>>>>>>>> Debezium >>>>>>>>>>>>>>>> Connector arrived in Flink 1.11.0, from looking at the >>>>>>>>>>>>>>>> documentation. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-flink.html >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/debezium.html >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> I'm wondering what alternative solutions are available for >>>>>>>>>>>>>>>> connecting Debezium to Flink? Is there an open source Debezium >>>>>>>>>>>>>>>> connector >>>>>>>>>>>>>>>> that works with Flink 1.10.0? Could I potentially pull the >>>>>>>>>>>>>>>> code out for the >>>>>>>>>>>>>>>> 1.11.0 Debezium connector and compile it in my project using >>>>>>>>>>>>>>>> Flink 1.10.0 >>>>>>>>>>>>>>>> api? >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> For context, I plan on doing some fairly complicated long >>>>>>>>>>>>>>>> lived stateful joins / materialization using the Table API >>>>>>>>>>>>>>>> over data >>>>>>>>>>>>>>>> ingested from Postgres and possibly MySQL. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Appreciate any help, thanks! >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> -- >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Rex Fenley | Software Engineer - Mobile and Backend >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Remind.com <https://www.remind.com/> | BLOG >>>>>>>>>>>>>>>> <http://blog.remind.com/> | FOLLOW US >>>>>>>>>>>>>>>> <https://twitter.com/remindhq> | LIKE US >>>>>>>>>>>>>>>> <https://www.facebook.com/remindhq> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> -- >>>>>>>>>>>>>> >>>>>>>>>>>>>> Rex Fenley | Software Engineer - Mobile and Backend >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> Remind.com <https://www.remind.com/> | BLOG >>>>>>>>>>>>>> <http://blog.remind.com/> | FOLLOW US >>>>>>>>>>>>>> <https://twitter.com/remindhq> | LIKE US >>>>>>>>>>>>>> <https://www.facebook.com/remindhq> >>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> -- >>>>>>>>>>>> >>>>>>>>>>>> Rex Fenley | Software Engineer - Mobile and Backend >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> Remind.com <https://www.remind.com/> | BLOG >>>>>>>>>>>> <http://blog.remind.com/> | FOLLOW US >>>>>>>>>>>> <https://twitter.com/remindhq> | LIKE US >>>>>>>>>>>> <https://www.facebook.com/remindhq> >>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>> >>>>>>>>>> -- >>>>>>>>>> >>>>>>>>>> Rex Fenley | Software Engineer - Mobile and Backend >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> Remind.com <https://www.remind.com/> | BLOG >>>>>>>>>> <http://blog.remind.com/> | FOLLOW US >>>>>>>>>> <https://twitter.com/remindhq> | LIKE US >>>>>>>>>> <https://www.facebook.com/remindhq> >>>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -- >>>>>>>>> >>>>>>>>> Rex Fenley | Software Engineer - Mobile and Backend >>>>>>>>> >>>>>>>>> >>>>>>>>> Remind.com <https://www.remind.com/> | BLOG >>>>>>>>> <http://blog.remind.com/> | FOLLOW US >>>>>>>>> <https://twitter.com/remindhq> | LIKE US >>>>>>>>> <https://www.facebook.com/remindhq> >>>>>>>>> >>>>>>>> >>>>>>> >>>>>>> -- >>>>>>> >>>>>>> Rex Fenley | Software Engineer - Mobile and Backend >>>>>>> >>>>>>> >>>>>>> Remind.com <https://www.remind.com/> | BLOG >>>>>>> <http://blog.remind.com/> | FOLLOW US >>>>>>> <https://twitter.com/remindhq> | LIKE US >>>>>>> <https://www.facebook.com/remindhq> >>>>>>> >>>>>> >>>>>> >>>>>> -- >>>>>> >>>>>> Arvid Heise | Senior Java Developer >>>>>> >>>>>> <https://www.ververica.com/> >>>>>> >>>>>> Follow us @VervericaData >>>>>> >>>>>> -- >>>>>> >>>>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink >>>>>> Conference >>>>>> >>>>>> Stream Processing | Event Driven | Real Time >>>>>> >>>>>> -- >>>>>> >>>>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany >>>>>> >>>>>> -- >>>>>> Ververica GmbH >>>>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B >>>>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, >>>>>> Ji (Toni) Cheng >>>>>> >>>>> >>>>> >>>>> -- >>>>> >>>>> Rex Fenley | Software Engineer - Mobile and Backend >>>>> >>>>> >>>>> Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/> >>>>> | FOLLOW US <https://twitter.com/remindhq> | LIKE US >>>>> <https://www.facebook.com/remindhq> >>>>> >>>> >>>> >>>> -- >>>> >>>> Arvid Heise | Senior Java Developer >>>> >>>> <https://www.ververica.com/> >>>> >>>> Follow us @VervericaData >>>> >>>> -- >>>> >>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink >>>> Conference >>>> >>>> Stream Processing | Event Driven | Real Time >>>> >>>> -- >>>> >>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany >>>> >>>> -- >>>> Ververica GmbH >>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B >>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji >>>> (Toni) Cheng >>>> >>> >>> >>> -- >>> >>> Rex Fenley | Software Engineer - Mobile and Backend >>> >>> >>> Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/> >>> | FOLLOW US <https://twitter.com/remindhq> | LIKE US >>> <https://www.facebook.com/remindhq> >>> >> >> >> -- >> >> Rex Fenley | Software Engineer - Mobile and Backend >> >> >> Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/> >> | FOLLOW US <https://twitter.com/remindhq> | LIKE US >> <https://www.facebook.com/remindhq> >> > -- Rex Fenley | Software Engineer - Mobile and Backend Remind.com <https://www.remind.com/> | BLOG <http://blog.remind.com/> | FOLLOW US <https://twitter.com/remindhq> | LIKE US <https://www.facebook.com/remindhq>
