Batch version of StreamingFileSink.forRowFormat(...)

Hi. I have a streaming job that writes to StreamingFileSink.forRowFormat(...) with an encoder that converts protocol buffers to byte arrays. How do read this data back in during a batch pipeline (using DataSet)? Do I use env.readFile with a custom DelimitedInputFormat? The streamfile sink docume

Issues with Flink Batch and Hadoop dependency

I'm assuming I have a simple, common setup problem. I've spent 6 hours debugging and haven't been able to figure it out. Any help would be greatly appreciated. *Problem* I have a Flink Streaming job setup that writes SequenceFiles in S3. When I try to create a Flink Batch job to read these Seq

Re: Batch version of StreamingFileSink.forRowFormat(...)

projects/flink/flink-docs-master/dev/table/connectors/filesystem.html > [2] > > https://ci.apache.org/projects/flink/flink-docs-master/dev/table/hive/hive_read_write.html > > On 11.08.20 00:13, Dan Hill wrote: > > Hi. I have a streaming job that writes to > > StreamingFileSink.for

Re: Issues with Flink Batch and Hadoop dependency

I was able to get a basic version to work by including a bunch of hadoop and s3 dependencies in the job jar and hacking in some hadoop config values. It's probably not optimal but it looks like I'm unblocked. On Fri, Aug 28, 2020 at 12:11 PM Dan Hill wrote: > I'm assumi

Flink Table API and not recognizing s3 plugins

*Background* I'm converting some prototype Flink v1.11.1 code that uses DataSet/DataTable APIs to use the Table API. *Problem* When switching to using the Table API, my s3 plugins stopped working. I don't know why. I've added the required maven table dependencies to the job. I've tried us movin

Re: Flink Table API and not recognizing s3 plugins

vided by flink-s3-fs-presto plugin. The plugin you used > provides s3a. > (and both provide s3, but it's good to use the more specific prefix). > > Best, > > Arvid > > On Thu, Sep 10, 2020 at 9:24 AM Dan Hill wrote: > >> *Background* >> I'm converting so

Re: Flink Table API and not recognizing s3 plugins

> *Usually by adjusting FLINK_HOME/conf/log4j.properties. > > On Thu, Sep 10, 2020 at 5:38 PM Dan Hill wrote: > >> Ah, sorry, it's a copy/paste issue with this email. I've tried both: >> 1) using s3a uri with flink-s3-fs-hadoop jar >> in /opt/flink/plugins

Re: Flink Table API and not recognizing s3 plugins

t 10:50 AM Dan Hill wrote: > I changed the levels to DEBUG. I don't see useful data in the logs. > > > https://drive.google.com/file/d/1ua1zsr3BInY_8xdsWwA__F0uloAqy-vG/view?usp=sharing > > On Thu, Sep 10, 2020 at 8:45 AM Arvid Heise wrote: > >> Could you try 1) or

Re: Flink Table API and not recognizing s3 plugins

s > neither on jobmanager nor on taskmanager. It occurs wherever you execute > the command line interface. > > How do you execute the job? Do you start it from your machine? Can you try > out to also add the respective s3 plugin there? > > Best, > > Arvid > > On Th

Re: Flink Table API and not recognizing s3 plugins

are pretty embedded into the Helm chart. My log4j-cli.properties is hacked on the CLI side. I thought I just needed the s3 plugins in the jobmanager and taskmanager. Do I need to have a similar plugin structure from the image where I run 'flink run'? On Thu, Sep 10, 2020 at 1:03 PM Dan Hi

Re: Flink Table API and not recognizing s3 plugins

This is running on my local minikube and is trying to hit minio. On Thu, Sep 10, 2020 at 1:10 PM Dan Hill wrote: > I'm using this Helm chart > <https://github.com/riskfocus/helm-charts-public/tree/master/flink>. I > start the job by building an image with the job jar and

I hit a bad jobmanager address when trying to use Flink SQL Client

I just tried using the Flink SQL Client. A simple job is not running because it cannot hit jobmanager. I'm not sure why Flink SQL Client is hitting "flink-jobmanager/10.98.253.58:8081". I'd expect either "flink-jobmanager:8081" or "10.98.253.58:8081" (which should work with my kubernetes setup).

Re: I hit a bad jobmanager address when trying to use Flink SQL Client

es Session? > Afaik this is not a documented, tested and officially supported feature > (this doesn't mean we should not support it -- apparently it is something > we should do rather soon ;) ). > > Best, > Robert > > On Fri, Sep 11, 2020 at 5:25 AM Dan Hill wrote: > >&

Re: Flink Table API and not recognizing s3 plugins

directories from the client. I think it might be problematic, as >> your case shows. We might need to revisit that part. I am cc'ing Jingsong >> who worked on the FileSystemSink. >> >> As a workaround you might try putting the s3 plugin on the CLI classpath >> (not

Re: Flink Table API and not recognizing s3 plugins

in Flink 1.10. > > The official Flink docker image has an easy way to add these plugins [1]. > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/ops/deployment/docker.html#using-plugins > > On Tue, Sep 15, 2020 at 6:40 AM Dan Hill wrote: > >> Thanks for the up

Re: Flink Table API and not recognizing s3 plugins

Sweet, this was the issue. I got this to work by copying the s3 jar over to plugins for the client container. Thanks for all of the help! The Table API is sweet! On Mon, Sep 14, 2020 at 11:14 PM Dan Hill wrote: > Yes, the client runs in K8. It uses a different K8 config than the H

Flink Table SQL, Kafka, partitions and unnecessary shuffling

How do I avoid unnecessary reshuffles when using Kafka as input? My keys in Kafka are ~userId. The first few stages do joins that are usually (userId, someOtherKeyId). It makes sense for these joins to stay on the same machine and avoid unnecessary shuffling. What's the best way to avoid unnece

Error: Avro "CREATE TABLE" with nested rows and missing fields

I might be misunderstanding Flink Avro support. I assumed not including a field in "CREATE TABLE" would work fine. If I leave out any field before a nested row, "CREATE TABLE" fails. If I include all of the fields, this succeeds. I assumed fields would be optional. I'm using Flink v1.11.1 with

Re: Flink Table SQL, Kafka, partitions and unnecessary shuffling

sure if and how can it be integrated > with the Table API. Maybe someone more familiar with the Table API can help > with that? > > Piotrek > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/experimental.html#reinterpreting-a-pre-partitioned-data-st

Re: Error: Avro "CREATE TABLE" with nested rows and missing fields

k it is a valid scenario to support such a case for the > combination of avro + filesystem. Honestly we do that already for > avro-schema-registry format (where we look up the writer schema in SR and > convert to the schema of DDL). Moreover it should be relatively easy to do > that.

Re: Flink Table SQL, Kafka, partitions and unnecessary shuffling

pache.org/projects/flink/flink-docs-release-1.11/dev/table/common.html#convert-a-datastream-or-dataset-into-a-table > On 16/09/2020 18:22, Dan Hill wrote: > > Hi Piotr! Yes, that's what I'm using with DataStream. It works well in > my prototype. > > On Wed, Sep 16, 2020 at 8:

Re: Error: Avro "CREATE TABLE" with nested rows and missing fields

files we could/should probably use the schema from the > file. > > Best, > > Dawid > > > [1] https://issues.apache.org/jira/browse/FLINK-16048 > On 16/09/2020 21:20, Dan Hill wrote: > > Interesting. How does schema evolution work with Avro and Flink? E.g.

Re: Flink Table SQL, Kafka, partitions and unnecessary shuffling

> > [1] > https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/queries.html#joins > > Best, > Godfrey > > > Dan Hill 于2020年9月17日周四 上午3:44写道： > >> Hi Dawid! >> >> I see. Yea, this would break my job after I move away from the

Re: I hit a bad jobmanager address when trying to use Flink SQL Client

14, 2020 at 2:18 PM Arvid Heise wrote: > >> Hi Dan, >> >> Can you verify from the pod that jobmanager and *10.98.253.58:8081 >> <http://10.98.253.58:8081> *is actually accessible (e.g., with curl)? >> I'd probably also try out localhost:8081 as you

Flink Table SQL and writing nested Avro files

Hi! I want to join two tables and write the results to Avro where the left and right rows are nested in the avro output. Is it possible to do this with the SQL interface? Thanks! - Dan CREATE TABLE `flat_avro` ( `left` ROW, `right` ROW ) WITH ( 'connector' = 'filesystem', 'path' =

Flink SQL - can I have multiple outputs per job?

I have a few results that I want to produce. - A join B - A join B join C - A join B join C join D - A join B join C join D join E When I use the DataSet API directly, I can execute all of these in the same job to reduce redundancy. When I use the SQL interface, it looks like separate jobs are cr

Re: Flink SQL - can I have multiple outputs per job?

I figured it out. TableEnvironment.StatementSet. Semi-related, query optimizers can mess up the reuse depending on which tables the join IDs come from. On Fri, Sep 18, 2020 at 9:40 PM Dan Hill wrote: > I have a few results that I want to produce. > - A join B > - A join B joi

Flink Table SQL and Job Names

I'm getting names like "insert-into_{catalog}.{db}.{table}". Is there a way to customize this with non-deprecated methods?

How to stop multiple Flink jobs of the same name from being created?

I'm prototyping with Flink SQL. I'm iterating on a client job with multiple INSERT INTOs. Whenever I have an error, my Kubernetes job retries. This creates multiple stream jobs with the same names. Is it up to clients to delete the existing jobs? I see Flink CLI functions for this. Do most pe

Re: How to stop multiple Flink jobs of the same name from being created?

client jars. On Sun, Sep 20, 2020 at 5:13 PM Dan Hill wrote: > I'm prototyping with Flink SQL. I'm iterating on a client job with > multiple INSERT INTOs. Whenever I have an error, my Kubernetes job > retries. This creates multiple stream jobs with the same names. > &g

Re: How to stop multiple Flink jobs of the same name from being created?

he.org/projects/flink/flink-docs-master/ops/deployment/kubernetes.html > [2]. > https://ci.apache.org/projects/flink/flink-docs-master/ops/deployment/native_kubernetes.html > > > Best, > Yang > > Dan Hill 于2020年9月21日周一 上午8:15写道： > >> I've read the following upgrade applicatio

Re: Flink Table SQL and writing nested Avro files

hema().getFieldCount() + 1, > t1.getSchema().getFieldCount() + > t2.getSchema().getFieldCount( > ) > .executeInsert("flat_avro") > .await(); > > > Best, > > Dawid > > [1] > https://ci.apache.

Back pressure with multiple joins

Hi! My goal is to better understand how my code impacts streaming throughput. I have a streaming job where I join multiple tables (A, B, C, D) using interval joins. Case 1) If I have 3 joins in the same query, I don't hit back pressure. SELECT ... FROM A LEFT JOIN B ON... LEFT JOIN C ON... LEFT

Re: Back pressure with multiple joins

When I use DataStream and implement the join myself, I can get 50x the throughput. I assume I'm doing something wrong with Flink's Table API and SQL interface. On Tue, Sep 22, 2020 at 11:21 PM Dan Hill wrote: > Hi! > > My goal is to better understand how my code impacts s

Reusing Flink SQL Client's Environment for Flink pipelines

Has anyone tried to reused the Flink SQL Client's yaml Environment configuration for their production setups? It seems pr

Best way to resolve bottlenecks with Flink?

My job has very slow throughput. What are the best signals that will indicate if there are performance issues? Is there an overall health summary that would indicate the most likely issues impacting performance? I found a variety of pages and metrics. I resolved some of the backpressure in my j

Re: Back pressure with multiple joins

> > > org.apache.flink.table.api.config.OptimizerConfigOptions#TABLE_OPTIMIZER_JOIN_REORDER_ENABLED > > Regards, > Timo > > On 23.09.20 08:23, Dan Hill wrote: > > When I use DataStream and implement the join myself, I can get 50x the > > throughput. I assume I'm do

Hiring Flink developers

I'm looking to hire Flink developers (full time or contractors) to work on a specialized user event logging system. Besides for the usual developer hiring websites, what are good hiring sources for Flink developers? Thanks! - Dan

Re: Flink SQL - can I force the planner to reuse a temporary view to be reused across multiple queries that access different fields?

> someone else can help here?). > > > > Piotrek > > > > [1] > > > https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/common.html#integration-with-datastream-and-dataset-api > > > > sob., 26 wrz 2020 o 00:02 Dan Hill > <mailt

Issue with Flink - unrelated executeSql causes other executeSqls to fail.

I'm writing a test for a batch job using MiniClusterResourceConfiguration. Here's a simple description of my working test case: 1) I use TableEnvironment.executeSql(...) to create a source and sink table using tmp filesystem directory. 2) I use executeSql to insert some test data into the source t

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

Could you enable > logging and see from the logs whether the 2) and 3) jobs execute > correctly on the MiniCluster? > > Best, > Aljoscha > > On 06.10.20 08:08, Dan Hill wrote: > > I'm writing a test for a batch job using > MiniClusterResourceConfigura

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

n implement a custom parallel source function for emitting watermarks if your job uses event time timers." On Tue, Oct 6, 2020 at 2:42 PM Dan Hill wrote: > I've tried to enable additional logging for a few hours today. I think > something with junit5 is swallowing the logs. I

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

Bazel to run Flink SQL tests on a mini cluster and >>> haven’t had issues, though we’re only testing on streaming jobs. >>> >>> Happy to help setting up logging with that if you’d like. >>> >>> Best, >>> Austin >>> >>> On Tue,

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

nstead of Kafka). - What's a good way of forcing a watermark using the Table API? On Tue, Oct 6, 2020 at 3:35 PM Dan Hill wrote: > Thanks! > > Great to know. I copied this junit5-jupiter-starter-bazel > <https://github.com/junit-team/junit5-samples/tree/main/junit5-jupit

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

"//src/main/resources:log4j.properties"],". Of >> course you also need to have the right log4j deps (or slf4j if you're >> using that) >> >> Hope that helps! >> >> Aljoscha >> >> On 07.10.20 00:41, Dan Hill wrote: >> > I

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

Switching to junit4 did not help. If I make a request to the url returned from MiniClusterWithClientResource.flinkCluster.getClusterClient().getWebInterfaceURL(), I get {"errors":["Not found."]}. I'm not sure if this is intentional. On Tue, Oct 6, 2020 a

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

On Wed, Oct 7, 2020 at 8:28 PM Dan Hill wrote: > Switching to junit4 did not help. > > If I make a request to the url returned from > MiniClusterWithClientResource.flinkCluster.getClusterClient().getWebInterfaceURL(), > I get > {"errors":["Not found."]}.

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

t5 w/ the resource files approach: > https://github.com/fintechstudios/vp-operator-demo-ff-virtual-2020/tree/master/tools/junit > > Best, > Austin > > On Thu, Oct 8, 2020 at 2:41 AM Dan Hill wrote: > >> I was able to get finer grained logs showing. I switched from >>

Any testing issues when using StreamTableEnvironment.createTemporaryView?

*Summary* I'm hitting an error when running a test that is related to using createTemporaryView to convert a Protobuf input stream to Flink Table API. I'm not sure how to debug "SourceConversion$5.processElement(Unknown Source)" line. Is this generated code? How can I debug this? Any help would

Re: Any testing issues when using StreamTableEnvironment.createTemporaryView?

I figured out my issue. I needed to assign watermarks (e.g. assignTimestampsAndWatermarks) after the fromElements. I could not figure out how the auto-generated code worked. I hooked up a debugger and guessed at the issue. On Thu, Oct 8, 2020 at 11:09 PM Dan Hill wrote: > *Summary* &g

Re: Issue with Flink - unrelated executeSql causes other executeSqls to fail.

on? You can get a CompletableFuture for the JobResult > which should help you. > > Best, > Aljoscha > > On 08.10.20 23:55, Dan Hill wrote: > > I figured out the issue. The join caused part of the job's execution to > be > > delayed. I added my own hacky wait condition int

Table SQL and updating rows

*Context* I'm working on a user event logging system using Flink. I'm tracking some info that belongs to the user's current session (e.g. location, device, experiment info). I don't have a v1 requirement to support mutability but I want to plan for it. I think the most likely reason for mutatio

Flink Kubernetes / Helm

What libraries do people use for running Flink on Kubernetes? Some links I've found: - Flink official documentation - Ververica documentation

Trying to run Flink tests

I forked Flink to work on a PR. When I run `mvn clean package` from a clean branch, Maven says the runtime tests failed but the logs do not appear to have details on the failure. Do I have to do anything to run these? ... [INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed:

Re: Trying to run Flink tests

R. > > If it is intended to execute the tests locally, you can try the following > actions. I'm not sure whether that helps though. > - Try to add '-DfailIfNoTests=false' to your maven command. > - Execute the maven command with '-X' to print all the debug

Re: Trying to run Flink tests

Song wrote: > Would you be able to share the complete maven logs and the command? And > what is the maven version? > > Thank you~ > > Xintong Song > > > > On Wed, Oct 21, 2020 at 1:37 AM Dan Hill wrote: > >> Hi Xintong! >> >> No changes. I tried -X a

Re: Trying to run Flink tests

the maven > logs. > - Quick question: which PR are you working on? By any chance you called > `System.exit()` in your codes? > > Thank you~ > > Xintong Song > > > > On Thu, Oct 22, 2020 at 5:59 AM Dan Hill wrote: > >> Sure, here's a link >> <ht

Flink Table SQL and MongoDB connector?

Has anyone connected these two? Looking through previous emails and the Flink docs, I've see two mentions of how to hook up MongoDB to Flink. 1) https://github.com/okkam-it/flink-mongodb-test 2) Debezium->Kafka->Flink https://debezium.io/documentation/reference/1.3/connectors/mongodb.html The de

Re: Trying to run Flink tests

3. > I'm not sure whether the maven version is related, but maybe you can try > it out with 3.2.5. And if it turns out worked, we may fire a issue at the > Apache Maven community. > > Thank you~ > > Xintong Song > > > > On Thu, Oct 22, 2020 at 12:31 PM Dan Hill

How to debug a Flink Exception that does not have a stack trace?

In the Flink dashboard, my job is failing with a NullPointerException but the Exception is not showing a stack trace. I do not see any NullPointerExceptions in any of the flink-jobmanager and flink-taskmanager logs. Is this a normal issue? [image: Screen Shot 2020-12-09 at 4.29.30 PM.png]

Re: How to debug a Flink Exception that does not have a stack trace?

Wed, Dec 9, 2020 at 4:33 PM Dan Hill wrote: > In the Flink dashboard, my job is failing with a NullPointerException but > the Exception is not showing a stack trace. I do not see any > NullPointerExceptions in any of the flink-jobmanager and flink-taskmanager > logs. > > Is thi

Re: How to debug a Flink Exception that does not have a stack trace?

t; { > IndexedRecord record = (IndexedRecord) avroObject; > GenericRowData row = new GenericRowData(arity); > for (int i = 0; i < arity; ++i) { > row.setField(i, fieldConverters[i].convert(record.get(i))); > } > return row; > }; >

Re: How to debug a Flink Exception that does not have a stack trace?

dn't reproduce it locally with a simple example job. > > Cheers, > Till > > On Thu, Dec 10, 2020 at 5:51 PM Dan Hill wrote: > >> Yea, the error makes sense and was an easy fix. >> >> Any idea what happened with the hidden stacktrace? The hidden stacktrace &g

Re: How to debug a Flink Exception that does not have a stack trace?

ce in the log > before the optimization kicks in. > > On Sat, Dec 12, 2020 at 2:05 AM Till Rohrmann > wrote: > >> Ok, then let's see whether it reoccurs. What you could do is to revert >> the fix and check the stack trace again. >> >> Cheers, >

Flink - Create Temporary View and "Rowtime attributes must not be in the input rows of a regular join"

When I try to refactor my joins into a temporary view to share joins and state, I get the following error. I tried a few variations of the code snippets below (adding TIMESTAMP casts based on Google searches). I removed a bunch of fields to simplify this example. Is this a known issue? Do I hav

Flink - sending clicks+impressions to AWS Personalize

I want to try using AWS Personalize to get content recommendations. One of the fields on the input (click) event is a list of recent impressions. E.g. { ... eventType: 'click', eventId: 'click-1', itemId: 'item-1' impression: ['item-2', 'item-3', 'i

Flink SQL, temporal joins and backfilling data

Hi! I have a Flink SQL job that does a few temporal joins and has been running for over a month on regular data. No issues. Ran well. I'm trying to re-run the Flink SQL job on the same data set but it's failing to checkpoint and very slow to make progress. I've modified some of the checkpoint

Flink SQL - IntervalJoin doesn't support consuming update and delete - trying to deduplicate rows

Hi! I'm using Flink SQL to do an interval join. Rows in one of the tables are not unique. I'm fine using either the first or last row. When I try to deduplicate and then interval join, I get

Comparing Flink vs Materialize

Has anyone compared Flink with Materialize? A friend recommended me switch to Materialize. In one of their blog posts, it says that Flink splits operators across CPUs (instead of splitting partitions across CPUs). Is this true? Is it configurable? https://materialize.com/blog-rocksdb/

Re: Flink SQL - IntervalJoin doesn't support consuming update and delete - trying to deduplicate rows

timization of deduplication in 1.12, which would reduce the required >> state tremendously. >> I'm pulling in Jark who knows more. >> >> On Thu, Dec 31, 2020 at 6:54 AM Dan Hill wrote: >> >>> Hi! >>> >>> I'm using Flink SQL to do an

Re: Flink SQL, temporal joins and backfilling data

ime or processing-time?) and checkpoint > settings? > > Regards, > Timo > > On 30.12.20 08:30, Dan Hill wrote: > > Hi! > > > > I have a Flink SQL job that does a few temporal joins and has been > > running for over a month on regular data. No issues. Ran w

Flink SQL and checkpoints and savepoints

How well does Flink SQL work with checkpoints and savepoints? I tried to find documentation for it in v1.11 but couldn't find it. E.g. what happens if the Flink SQL is modified between releases? New columns? Change columns? Adding joins?

Re: Flink SQL and checkpoints and savepoints

nds on the used operators. > > Regards, > Timo > > > On 18.01.21 04:46, Dan Hill wrote: > > How well does Flink SQL work with checkpoints and savepoints? I tried > > to find documentation for it in v1.11 but couldn't find it. > > > > E.g. what happens if t

Re: How to safely update jobs in-flight using Apache Flink on AWS EMR?

Hi! I'm assuming this question comes up regularly with AWS EMR. I posted it on Stack Overflow. https://stackoverflow.com/questions/62309400/how-to-safely-update-jobs-in-flight-using-apache-flink-on-aws-emr I was not able to find instructions for

Does anyone have an example of Bazel working with Flink?

I took the Flink playground and I'm trying to swap out Maven for Bazel. I got to the point where I'm hitting the following error. I want to diff my code with an existing, working setup. Thanks! - Dan client_1| org.apache.flink.client.program.ProgramInvocationException: Neither

Re: Does anyone have an example of Bazel working with Flink?

t >>> or dependency my application needs. >>> >>> Unfortunately I'm not sure what issue you're seeing here. I would guess >>> either your flink application wasn't compiled into the jar >>> you're executing. If you can paste the bazel target

Re: Flink SQL and checkpoints and savepoints

g/confluence/display/FLINK/FLIP-150%3A+Introduce+Hybrid+Source > > Regards, > Timo > > > On 18.01.21 10:40, Dan Hill wrote: > > Thanks Timo! > > > > The reason makes sense. > > > > Do any of the techniques make it easy to support exactly once? > &g

Re: Flink SQL and checkpoints and savepoints

I went through a few of the recent Flink Forward videos and didn't see solutions to this problem. It sounds like some companies have solutions but they didn't talk about them in enough detail to do something similar. On Thu, Jan 28, 2021 at 11:45 PM Dan Hill wrote: > Is this savep

Minicluster Flink tests, checkpoints and inprogress part files

Hi Flink user group, *Background* I'm changing a Flink SQL job to use Datastream. I'm updating an existing Minicluster test in my code. It has a similar structure to other tests in flink-tests. I call StreamExecutionEnvironment.execute. My tests sink using StreamingFileSink Bulk Formats to tmp

Re: Minicluster Flink tests, checkpoints and inprogress part files

nt before > shutting down. This will allow them to move the last outstanding > inprogress files over to finished as well. > > [2] https://cwiki.apache.org/confluence/x/mw-ZCQ > > I hope that helps! > > Best, > Aljoscha > > On 2021/02/04 21:37, Dan Hill wrote: > >Hi

Re: Minicluster Flink tests, checkpoints and inprogress part files

Ah looks like I need to use 1.12 for this. I'm still on 1.11. On Fri, Feb 5, 2021, 08:37 Dan Hill wrote: > Thanks Aljoscha! > > On Fri, Feb 5, 2021 at 1:48 AM Aljoscha Krettek > wrote: > >> Hi Dan, >> >> I'm afraid this is not easily possible using t

Re: Minicluster Flink tests, checkpoints and inprogress part files

I changed the test to use ExecutionMode.BATCH in v1.11 and it still doesn't work. How did devs write minicluster tests before for similar code? Did they not? On Sat, Feb 6, 2021 at 5:38 PM Dan Hill wrote: > Ah looks like I need to use 1.12 for this. I'm still on 1.11. > >

UUID in part files

Hi. *Context* I'm migrating my Flink SQL job to DataStream. When switching to StreamingFileSink, I noticed that the part files now do not have a uuid in them. "part-0-0" vs "part-{uuid string}-0-0". This is easy to add with OutputFileConfig. *Question* Is there a reason why the base OutputFile

Optimizing Flink joins

Hi! I was curious if there are docs on how to optimize Flink joins. I looked around and on the Flink docs and didn't see much. I see a little on the Configuration page. E.g. one of my jobs has an interval join. Does left vs right matter for interval join?

Re: Optimizing Flink joins

c, b, a").explain()); > > So you can reorder the tables in the query if that improves performance. > For interval joins, we currently don't provide additional algorithms or > options. > > Regards, > Timo > > On 11.02.21 05:04, Dan Hill wrote: > > Hi! I was curio

Re: Optimizing Flink joins

wrote: > Hi Dan, > > thanks for letting us know. Could you give us some feedback what is > missing in SQL for this use case? Are you looking for some broadcast > joining or which kind of algorithm would help you? > > Regards, > Timo > > On 11.02.21 20:32, Dan Hill w

Best practices around checkpoint intervals and sizes?

Hi. I'm playing around with optimizing our checkpoint intervals and sizes. Are there any best practices around this? I have a ~7 sequential joins and a few sinks. I'm curious what would result in the better throughput and latency trade offs. I'd assume less frequent checkpointing would increas

Debugging long Flink checkpoint durations

Hi. Are there good ways to debug long Flink checkpoint durations? I'm running a backfill job that runs ~10 days of data and then starts checkpointing failing. Since I only see the last 10 checkpoints in the jobmaster UI, I don't see when it starts. I looked through the text logs and didn't see

Re: Debugging long Flink checkpoint durations

Thanks! Yes, I've looked at these. My job is facing backpressure starting at an early join step. I'm unclear if more time is fine for the backfill or if I need more resources. On Tue, Mar 2, 2021 at 12:50 AM Yun Gao wrote: > Hi Dan, > > I think you could see the detail of the checkpoints via

Flink, local development, finish processing a stream of Kafka data

Hi. For local and tests development, I want to flush the events in my system to make sure I'm processing everything. My watermark does not progress to finish all of the data. What's the best practice for local development or tests? If I use idle sources for 1 Kafka partition, this appears broke

Re: Debugging long Flink checkpoint durations

apache.org/projects/flink/flink-docs-release-1.11/ops/memory/mem_setup_tm.html On Tue, Mar 2, 2021 at 3:45 PM Dan Hill wrote: > Thanks! Yes, I've looked at these. My job is facing backpressure > starting at an early join step. I'm unclear if more time is fine for the >

Re: Debugging long Flink checkpoint durations

The checkpoint was only acknowledged shortly after it was started. On Thu, Mar 4, 2021 at 12:38 PM Dan Hill wrote: > I dove deeper into it and made a little more progress (by giving more > resources). > > Here is a screenshot of one bottleneck: > https://drive.go

Gradually increasing checkpoint size

Hi! I'm running a backfill Flink stream job over older data. It has multiple interval joins. I noticed my checkpoint is regularly gaining in size. I'd expect my checkpoints to stabilize and not grow. Is there a setting to prune useless data from the checkpoint? My top guess is that my checkpo

Re: Gradually increasing checkpoint size

k. > > Best, > Yun > > > --Original Mail -- > *Sender:*Dan Hill > *Send Date:*Mon Mar 8 14:59:48 2021 > *Recipients:*user > *Subject:*Gradually increasing checkpoint size > >> Hi! >> >> I'm running a backfill Flink

Re: Re: Gradually increasing checkpoint size

ttached operator has a `alreadyOutputed` value state, which seems to keep > increasing if there are always new keys ? > > Best, > Yun > > > --Original Mail -- > *Sender:*Dan Hill > *Send Date:*Tue Mar 9 00:59:24 2021 > *Recipient

Best practices for complex state manipulation

Hi! I'm working on a join setup that does fuzzy matching in case the client does not send enough parameters to join by a foreign key. There's a few ways I can store the state. I'm curious about best practices around this. I'm using rocksdb as the state storage. I was reading the code for Interv

Re: Best practices for complex state manipulation

ommend starting from this webinar > done by my colleague Seth Weismann: > https://www.youtube.com/watch?v=9GF8Hwqzwnk. > > Cheers, > Gordon > > On Wed, Mar 10, 2021 at 1:58 AM Dan Hill wrote: > >> Hi! >> >> I'm working on a join setup that does fuzzy mat

Re: Gradually increasing checkpoint size

7;t think it's where you pointed. > I'd suggest to first investigate the progress of watermarks. > > Best, > > Dawid > On 09/03/2021 08:36, Dan Hill wrote: > > Hi Yun! > > That advice was useful. The state for that operator is very small > (31kb). Most o

Does WatermarkStrategy.withIdleness work?

I haven't been able to get WatermarkStrategy.withIdleness to work. Is it broken? None of my timers trigger when I'd expect idleness to take over. On Tue, Mar 2, 2021 at 11:15 PM Dan Hill wrote: > Hi. > > For local and tests development, I want to flush the events in my sy

Re: Does WatermarkStrategy.withIdleness work?

int > [2] > https://github.com/aljoscha/flink/blob/6e4419e550caa0e5b162bc0d2ccc43f6b0b3860f/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/timestamps/ProcessingTimeTrailingBoundedOutOfOrdernessTimestampExtractor.java > > On Fri, Mar 12, 2021 at 9:47 AM D

  1   2   >