https://flink.apache.org/ logo
Join Slack
Powered by
Has anyone here been able to perform upgrades with...
# troubleshooting
v
Has anyone here been able to perform upgrades with savepoints when using Flink Table API? I end up with:
Copy code
Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint <s3://savepoints/gsp/savepoint-d9ef6a-babcec0db851>. Cannot map checkpoint/savepoint state for operator 8a238689f724f6c1740b0a7bf83ec304 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
I have:
Copy code
table.exec.uid.generation: "ALWAYS"
    table.exec.legacy-transformation-uids: "true"
I also have 
.disableAutoGeneratedUIDs()
to ensure UIDs are explicitly set. Why is the ID not following the default format 
"<id>_<transformation>"
also? From https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/config/#table-exec-uid-format
πŸ‘€ 1
Is the only way through compiled plans?
m
Is the only way through compiled plans?
Yes
v
Right, but if I need to, for example, reduce the parallelism, wouldn't that generate a new plan with new IDs?
So a upgrade isn't really possible?
m
Where do you set the parallelism?
Via the 
SET
option?
v
Copy code
tableEnv.getConfig.getConfiguration
      .setInteger("table.exec.resource.default-parallelism", jobSettings.tableParallelism)
but the source has different parallelism, so we use 
kafkaSource.setParallelism(settings.sourceParallelism)
. most of the times they are the same however
m
That's a DataStream operation, not a SQL operation
v
yes, we are using both
m
And there's no support for DataStream & Table API upgrades
Also not with CompiledPlans
At least, not yet IIRC
But would have to double check the FLIP
v
oversimplification but the pipeline is more or less 
kafkaSource -> convert to tables via .fromChangelogStream table api (joins) -> convert back to datastreams via .toChangelogStream -> windows -> kafkaSink
Right so compiled plans are only for the table api stuff?
m
Why wouldn't you use the Table Source and Table Sink?
Right so compiled plans are only for the table api stuff?
For anything in the SQL and Table API, but not 100% implemented yet
v
because in one job we have n amount of pipelines, that would mean n amount of kafka producers as well. sometimes the number is quite high, > 100
m
But then you would use a Statement Set, right?
v
We have one Kafka topic containing Debezium CDC data from multiple MSSQL tables. Right now we are using the Datastream Kafka connector + side outputs to create a changelog stream for each table. so
Copy code
debeziumData (tableA, tableB) 
  ~> tableA (sideOutput) ~> .fromChangelogStream ~> tableEnv.createTemporaryView (Registers tableA in the catalog)
  ~> tableB (sideOutput) ~> .fromChangelogStream ~> tableEnv.createTemporaryView (Registers tableB in the catalog)
then we are essentially materializing MSSQL views. example of one:
Copy code
CREATE VIEW viewA AS SELECT
  a.col1,
  b.col2
FROM tableA
JOIN tableA on tableB [...]
we have over 700 views we need to materialize. each view is a pipeline. we use Table API to do that because well we just need to copy and paste the CREATE VIEW AS SELECT [...] definition essentially.
We were using StatementSets, works fine, but statement set forces us to insert a tableX into tableY, tableY would be the upsert-kafka table, that means every view would spawn a kafka producer unnecessarily πŸ˜•
so instead we are now doing
Copy code
.toChangelogStream(viewA).map(toKafkaProducerRecord) ~>

.toChangelogStream(viewB).map(toKafkaProducerRecord) ~>       .reduce((x, y) => x.union(y)).toSink(kafkaSink)

.toChangelogStream(viewC).map(toKafkaProducerRecord) ~>
FLIP-190 does seem to be exactly what i need. specifically "DataStream API Sources" and "Table with DataStream API Sink" sections. seems like they haven't yet been implemented however : /
m
Exactly. Just not there yet
2 Views
Open in Slack
PreviousNext
Powered by