Your approach looks rather good to me. In the version with querying for the JobStatus you must remember that there are such states as e.g. INITIALIZING, which just tells you that the job was submitted.
In 1.12 we introduced the TableResult#await method, which is a shortcut
over what you did in the last version of yours.
Best,
Dawid
On 08/12/2020 22:40, Dylan Forciea wrote:
>
> After a bit more playing around with it today, I figured out that what
> I needed to call was:
>
>
>
> statementSet.execute().getJobClient().get().getJobExecutionResult(getClass().getClassLoader()).get()
>
>
>
> The fact that getJobExecutionResult required a classloader is what
> threw me off. Since I’m using an application cluster, I didn’t really
> think about the fact that I would need to supply one. It might be nice
> to have a flavor of this that just uses the default system class loader.
>
>
>
> However, I’m still interested in if there is a better way to handle
> reloading a table or tables in active use than the series of steps
> that I went through, if anybody has any suggestions!
>
>
>
> Thanks,
>
> Dylan Forciea
>
>
>
> *From: *Dylan Forciea <dy...@oseberg.io>
> *Date: *Monday, December 7, 2020 at 5:33 PM
> *To: *"user@flink.apache.org" <user@flink.apache.org>
> *Subject: *Re: Batch loading into postgres database
>
>
>
> As a follow up – I’m trying to follow the approach I outlined below,
> and I’m having trouble figuring out how to perform the step of doing
> the delete/insert after the job is complete.
>
>
>
> I’ve tried adding a job listener, like so, but that doesn’t seem to
> ever get fired off:
>
>
>
> valstatementSet= streamTableEnv.createStatementSet()
>
> statementSet.addInsertSql("""
>
> INSERT INTO table1_staging
>
> SELECT * FROM table
>
> """)
>
>
>
> statementSet.addInsertSql("""
>
> INSERT INTO table2_staging
>
> SELECT * FROM table
>
> """)
>
>
>
> statementSet.addInsertSql("""
>
> INSERT INTO table3_staging
>
> SELECT * FROM table3
>
> """)
>
>
>
> streamEnv.registerJobListener(newJobListener() {
>
> overridedefonJobSubmitted(jobClient: JobClient, throwable:
> Throwable): Unit= {}
>
>
>
> overridedefonJobExecuted(result: JobExecutionResult, throwable:
> Throwable): Unit= {
>
> valtime= Option(result).map(_.getNetRuntime())
>
> if(throwable == null) {
>
> Log.info(s"Completed job successfully in $timemilliseconds")
>
> } else{
>
> Log.error(s"Unable to execute job successfully", throwable)
>
> }
>
> }
>
> })
>
> statementSet.execute()
>
>
>
> I tried the above with the execute before and after the register, but
> it doesn’t seem to fire in any case.
>
>
>
> I also tried this:
>
>
>
> Try(statementSet.execute().getJobClient().get().getJobStatus().join())
>
> .map { _ =>
>
> Log.info(s"Completed job successfully")
>
> }
>
> .recover {
>
> caset => {
>
> Log.error(s"Unable to execute job successfully", t)
>
> }
>
> }
>
>
>
> And this seems to have fired WAY before the job actually finished
> flowing all the data through. I tried both join and get on the job
> status CompleteableFuture
>
>
>
> Is there anything I’m missing as far as being able to tell when the
> job is complete? Again, this is Flink 1.11.2 that I’m running.
>
>
>
> Thanks,
>
> Dylan Forciea
>
>
>
> *From: *Dylan Forciea <dy...@oseberg.io>
> *Date: *Monday, December 7, 2020 at 8:04 AM
> *To: *"user@flink.apache.org" <user@flink.apache.org>
> *Subject: *Batch loading into postgres database
>
>
>
> I am setting up a Flink job that will reload a table in a postgres
> database using the Flink SQL functionality. I just wanted to make sure
> that given the current feature set I am going about this the correct
> way. I am currently using version 1.11.2, but plan on upgrading to
> 1.12 soon whenever it is finalized.
>
>
>
> I have setup a staging table and a final table in a postgres database.
> My plan is to have a Flink application that will truncate the contents
> of the staging table before the job begins using JDBC, run the job to
> completion, and then with JDBC delete/insert into the final table from
> the staging table in a transaction after the job completes.
>
>
>
> Is this the expected way to interact with postgres in a batch job like
> this? Or is there some functionality or method that I am missing?
>
>
>
> Regards,
>
> Dylan Forciae
>
signature.asc
Description: OpenPGP digital signature
