Guys, I'm consuming a pulsar topic using a PulsarS...
# randomm
Merijn van Mourik
12/19/2022, 7:24 PMGuys, I'm consuming a pulsar topic using a PulsarSource<String>, and do a bit processing. When I notice messages flowing in, the stream seems to be processed every 10 seconds, not continuously. Also I run the flink stream standalone, just from IntelliJ. I'm a bit lost, can anyone give me a clue why this happens?
flink 1
m
Martijn Visser
12/19/2022, 7:25 PMLet me ping some of the connector maintainers, because I think they can answer this one the best. They are in a different timezone so replies might be later
๐ 3
m
Merijn van Mourik
12/19/2022, 7:27 PMThanks for that. I also grokked all logs, but could not find any interesting. First I tought maybe the DataStream.print() caused it. So turned that off too.
m
Martijn Visser
12/19/2022, 7:27 PMDo you have checkpointing enabled?
m
Merijn van Mourik
12/19/2022, 7:28 PMSeems to be turned off. I played with that feature in the past...
m
Martijn Visser
12/19/2022, 7:29 PMReason why I'm asking is because with Kafka we know that users are surprised because usually lag is measured by looking at the offsets, which in the case of Kafka are only committed when a checkpoint is made. Since Pulsar uses the same Flink APIs under the hood as a source, I thought this could be the same but I don't know Pulsar as a technology ๐
m
Merijn van Mourik
12/19/2022, 7:30 PMHmm, it might be interesting to turn that off explicitly. Good direction to search into... For you're information: I'm processing a live stream of 'matrixborden' of the Dutch highways๐
m
Martijn Visser
12/19/2022, 7:31 PMI figured with your name you must be a fellow-Dutchie ๐ that is a cool use case
Martijn Visser
12/19/2022, 7:31 PMKeep in mind that if you turn checkpointing off, there's no resilience pattern in case Flink crashes or needs to be restarted. But since you're running locally that won't be an issue now
m
Merijn van Mourik
12/19/2022, 7:32 PMThanks for that. Yes, I take some steps for streams to be recoverable. But for now it's not a big deal. Later on we will incorporate live speed measurements. It's a lot of data.
Merijn van Mourik
12/19/2022, 7:33 PMMerijn van Mourik
12/19/2022, 7:34 PMWorks beautifully ๐
m
Martijn Visser
12/19/2022, 7:34 PMIt's really cool to see these type of use cases! ๐ฅณ
flink 1
m
Merijn van Mourik
12/19/2022, 7:35 PMWe do live detection, also running tile38 for collision detection. I'm planning to use flink for realtime geo processing. Where we can detect wheter a big truck runs into a vulnerable roadworker, and send warnings to the driver.
Merijn van Mourik
12/19/2022, 7:35 PMSo we take into account distance, speed, direction.
Merijn van Mourik
12/19/2022, 7:36 PMAnd live joins of geo data (like whether a user is inside a polygon).
Merijn van Mourik
12/19/2022, 7:36 PMSmall steps.... but getting there.
Merijn van Mourik
12/19/2022, 7:40 PMI know ArcGIS is working on realtime data too. But I think compared to Pulsar/Flink it's old tech....
m
Martijn Visser
12/19/2022, 7:40 PMWould it help your use case if FLINK-9219 was implemented?
m
Merijn van Mourik
12/19/2022, 7:41 PMI'm playing with Sedona too... yes this stuff really helps!
m
Martijn Visser
12/19/2022, 7:42 PMThat's good to know
m
Merijn van Mourik
12/19/2022, 7:45 PMSedona also has these ST_ functions... probably some overlap there.
Merijn van Mourik
12/19/2022, 7:47 PM Copy code
StreamExecutionEnvironment env = EnvironmentUtils.initEnvWithWebUI(true);
env.setParallelism(1);
EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
SedonaFlinkRegistrator.registerType(env);
SedonaFlinkRegistrator.registerFunc(tableEnv);Merijn van Mourik
12/19/2022, 7:48 PMAnd then you can run those ST_ functions on tables.
m
Martijn Visser
12/19/2022, 7:51 PMWondering if they also leverage Calcite under the hood
Martijn Visser
12/19/2022, 7:51 PMAnother thing to put on my list of things to look into during a couple of days off ๐
m
Merijn van Mourik
12/19/2022, 7:52 PMThe story of my live ๐ณ๐คฃ
y
Yufan Sheng
12/19/2022, 9:25 PMHi @Merijn van Mourik . This is a known issue and get tracked in https://issues.apache.org/jira/browse/FLINK-28870. I forget to submit the PR on Flink. But we have fixed it in StreamNativeโs fork. You can find the connector version for your Flink here: https://mvnrepository.com/artifact/io.streamnative.connectors/flink-connector-pulsar
๐ 2
Yufan Sheng
12/19/2022, 9:28 PMThis issue only occur on small incoming messages. Feel free to use it in Production with massive messages.
๐ 2
m
Merijn van Mourik
12/20/2022, 12:00 PMI can confirm it works now! Thanks!!
Merijn van Mourik
12/20/2022, 12:01 PMI also experimented a bit with the pulsar client configuratoin options. Learning something new everytime...
๐ 2
s
sairam yeturi
12/23/2022, 1:28 PMIs there a channel for understanding usecases on flink? I loved to read this thanks for sharing the usecase
m
Merijn van Mourik
12/24/2022, 11:44 AMCurrently this is the place to go: https://flink.apache.org/usecases.html
Merijn van Mourik
12/24/2022, 11:47 AMI'm also looking at realtime data warehousing, and ways to process changing dimensions for metadata.
๐ 1
Merijn van Mourik
01/03/2023, 9:46 PMWow, learning Flink is not for the faint-harted ๐
Merijn van Mourik
01/03/2023, 9:46 PMmessage has been deleted
Merijn van Mourik
01/03/2023, 9:47 PMGetting some real results now...