Hi, about 2-3 weeks back I had started a thread regarding flink batch jobs that use multiple pipelines. We are able to achieve that using the ways suggested by @Weijie Guo. Just a follow up question on that. Just to give a context. I have around 30 tables sitting in postgresql database. I am running a loop. And for every iteration in that loop, I am 1. Fetching data from source postgresql table 2. Doing aggregations 3. Storing results in destination postgresql table 4. Running the execute command. But what we are observing is that there are 30 pipelines and every pipeline runs after one another. Issue is that the first pipeline gets finished in 90 seconds but it takes same amount of time for subsequent pipelines as well. So it takes almost around 30 pipelines * 90 seconds = 45 minutes to finish this job. Is this a limitation of Apache Flink? Or I am missing something

I don't fully understand your problem. If you want to execute all pipelines in sequence, the total execution time should be

T (pipeline1) + T(pipeline2) + T (pipeline30)

, do you mean that the actual time consuming is much more than this?

Sorry to bother. I think you are right. Need to debug each pipeline to check why its taking time. I have one other question though. Lets say I am loading data from source table in postgresql through table apis and then aggregating and then storing the data in destination postgresql table. I have 20 to 30 tables for which I need to do this operation ok. As I have mentioned, as per our earlier discussion, I was doing something like this for each table { 1. initialize streamExecution and streamTableEnvironemnt 2. Load data 3. aggregate 4. store data 5. streamExecutionEnvironment.execute() } This way it would run it sequentially. The only problem because of which I dont want to execute this in parallel is that if I execute it parallely like this Initialize streamExecEnv and streamTableEnv for each table { 1. Load data 2. aggregate 3. store data } streamExecutionEnv.execute() because then it would load the data all at once and create issue. But somehow in the table I am able to do something like load limited data say 10k rows, aggregate then pick up next 10k rows, aggregate etc then I can run everything simultaneously. is there any configuration in table api batch mode where I can set up the number of records to fetch. So that it wont load all data at once for all source tables??

is there any configuration in table api batch mode where I can set up the number of records to fetch. So that it wont load all data at once for all source tables?

This depends on which connector you use. To be honest, I don't particularly understand why reading partial data at one time can solve the problem. What I understand is: The reason why we must be executed in sequence is that there is data dependency. For example, the data of table2 is written after table1 is processed. Is the scene I understand not quite consistent with what you said？

Sometimes there is dependency. But somtimes the pipelines are independent

Ok, so you want source to read a batch of data at a time. For example, the interval is (from, end). I guess you use the JDBC connector, but I don't know it very well. And there seems to be no parameter in the

flink jdbc connector

document that can meet this behavior.🤔

Okk got it. But consider this use case (aggregation). And consider that all the tables and their aggregations are independent. Like there is no dependency in the pipeline. What is the best way to solve this use case? Whatever way that I have mentioned (multiple pipelines running one after another) is optimal???

For these tables, if you load all the data at one time, it will cause problems. Why? This is the root cause of the problem, and I need to know more about this part.

Because it depends on data size. If data is more then resources wont be sufficient and everything would be loaded in the memory simultaneously. So it would put pressure on memory. Correct me if I am wrong

Source does not put all data in memory at one time. It will read and process data batch by batch. In addition, for the Batch operator, there is generally not much data cached in memory.

Ohhh you mean even if I execute following code Initialize streamExecEnv and streamTableEnv for each table { 1. Load data 2. aggregate 3. store data } streamExecutionEnv.execute() It should not overwhelm the memory by loading all the data simultaneously???

Not all data will be loaded into memory at one time. If the memory is not sufficient in batch mode, it will generally affect performance. In addition, if you want to execute multiple jobs in parallel, it is necessary to consume more resources. After all, every job needs to be assigned TM. If you don't want to do this, maybe the session mode is a better choice.

Ok but when I try to run my program in parallel fashion it gives heap out of memory exception

Never mind. I need to ensure that which connector you use?

Actually I am using Table Apis. And in table APIs I am using catalogues

Ahh, and which catalog you use ? JDBC?

Yess JDBC catalog

the source (implemented using tabe api catalogues) only loads data through 1 thread and 3 of them remain idle

What exactly does

idle

mean? If I remember correctly, JDBC source has a partition number parameter. If the parallelism is greater than the partition number, the task corresponding to the exceeded part will be directly finished.

Ohhh yes thats what is happening

you can reference this doc to set some config option to enable parallel multiple tasks.https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/connectors/table/jdbc/#partitioned-scan

But I didnt get this. I am using something like this

val newCatalogue = new JdbcCatalog(catalogueName, dbName, username, password, urlWithoutDB)

Whatever options that you mentioned are from jdbc connector. I am using jdbc catalog. I am not sure how to pass these options in catalog

You can set this options by sql hint(https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/queries/hints/#dynamic-table-options)

Oh okk. Sure let me try this

Emm... To be honest, I'm not an expert of JDBC Connector, I just used it for a few times.

I was wondering, shouldnt it be a default behaviour of table APIs to do this partitioning internally if default parallelism is provided?

I guess this may be because Flink is hard to decide by which column it should partition.

Okk got it. No problem. Let me search up more. Would let you know if I find anything. Thanks a lot for the help. 🙂