https://flink.apache.org/ logo
Join Slack
Powered by
<@U03HM2QHSTH> I have seen your talk about “Storin...
# troubleshooting
t
@sap1ens I have seen your talk about “Storing State Forever: Why It Can Be Good For Your Analytics”. Very informative! I have a couple of questions about it: • If I am not mistaken, Flink CDC source operators require a parallelism of 1 to ensure that the sequence of update events is correct. Is it correct that for all the sources you used parallelism of 1 then, and for the sink a high parallelism to maximise resources for the join? • Any work in progress on backfilling the state? 🙂
s
Hi! 🙂 Thanks for the kind words.
If I am not mistaken, Flink CDC source operators require a parallelism of 1 to ensure that the sequence of update events is correct. Is it correct that for all the sources you used parallelism of 1 then, and for the sink a high parallelism to maximise resources for the join?
In the talk I didn’t use Flink CDC. We used Debezium with Kafka Connect and Flink was reading from Kafka topics. You’d only use parallelism of 1 if you want to keep global ordering. Usually you’re fine with correct ordering within a primary key. In this case you can call a 
keyBy
right after the source to get the correct semantics (as long as Kafka topics use the same key).
But if you use Flink CDC you might be forced to use parallelism of 1, because that’s what it supports (e.g. when reading from a single Postgres replication slot). I wrote a bit more on that here.
Any work in progress on backfilling the state?
Nothing to report 🙂
t
Ah great read! I think I actually came across the issue that checkpoints are not allowed during snapshotting. I ran 
flink stop --savepointPath ...
and got an error that checkpointing failed. However, it seemed that creating a manual savepoint by 
flink savepoint ... ...
actually creates a savepoint without checkpoints failing. After that I was able to cancel the job and start from the savepoint again. Any idea why this is the case?
s
Not really, it should have failed IMO
r
Hi @sap1ens, I watched the talk as well. It was quite interesting and easy to follow. I noticed you used Scala, Do the streams you are doing a union have the same type? I'd like to replicate the non-temporal join(a simpler version) in Java and I'm wondering how the 
keyby
would work because both streams don't have the same type. For reference:
Copy code
DataStream <TypeA> StreamA = ...
DataStream <TypeB> StreamB = ...

DataStream<TypeC> JoinedStream = StreamA.union(StreamB)
                .keyBy(new KeySelector<A or B ?>, String>() {
                    @Override
                    public String getKey(<A or B ?> value) throws Exception {
                        return value.getSomeKey;
                    }
                })
                .process(new JoinFunction());
JoinFunction
would have to return a different type 
TypeC
s
Hey, thanks! Regarding your question, you can create another type that’s a union of TypeA and TypeB. And for that type you’d implement a 
getKey
method that extracts the correct key from TypeA or TypeB, depending on what’s defined. You can use that for 
keyBy
r
Thanks! I'll try this
TypeC
contains all the fields in 
TypeA
and 
TypeB
as well as the common key. The union operators throws an exception still because 
TypeA
and 
TypeB
are not of the same type. Please let me know if I've understood you correctly. I don't really understand this part - "And for that type you’d implement a 
getKey
method that extracts the correct key from TypeA or TypeB"
Copy code
DataStream <TypeA> StreamA = ...
DataStream <TypeB> StreamB = ...

DataStream<TypeC> JoinedStream = StreamA.union(StreamB)
                .keyBy(new TypeC>, String>() {
                    @Override
                    public String getKey(TypeC value) throws Exception {
                        return value.getSomeKey;
                    }
                })
                .process(new JoinFunction());
s
Copy code
case class Wrapper(
  typeA: Option[TypeA] = None,
  typeB: Option[TypeB] = None
) {
  def getKey: String = {
    // check if typeA is defined, then return typeA.key, otherwise return typeB.key
  }
}
something like this
Then you convert both input streams to DataStream[Wrapper]
👍 1
Open in Slack
PreviousNext
Powered by