Hi Marco, Sadly, I myself haven't ever used StateProcessorAPI either.
I thought, the documentation here [1] is rather straight forward to be used, but I never tried it myself. Also, I don' get what you mean with "tieing" DataSet and DataStream? For what I understand, the StateProcessorAPI is internally upon DataSets at the moment. So if you have a Streaming job (DataStream), you would run your custom state migration program before even starting the streaming job using the StateProcessor API and initialize your state as a DataSet. After all, initializing a state doesn't require you to have an infinite job running, but only a finite one (Thus batch/DataSet). Once that program finished execution, you would submit your streaming job starting from the written savepoint. I guess the API works with either HDFS or filesystem, but maybe someone who has already used the API might shed some more light for us here. Best regards Theo [1] [ https://ci.apache.org/projects/flink/flink-docs-stable/dev/libs/state_processor_api.html | https://ci.apache.org/projects/flink/flink-docs-stable/dev/libs/state_processor_api.html ] Von: "Marco Villalobos" <mvillalo...@kineteque.com> An: "Theo Diefenthal" <theo.diefent...@scoop-software.de> CC: "user" <user@flink.apache.org> Gesendet: Donnerstag, 6. August 2020 23:47:13 Betreff: Re: Two Queries and a Kafka Topic I am trying to use the State Processor API. Does that require HDFS or a filesystem? I wish there was a complete example that ties in both DataSet and DataStream API, and the State Processor API. So far I have not been able to get it to work. Does anybody know where I can find examples of these type of techniques? On Wed, Aug 5, 2020 at 3:52 AM Theo Diefenthal < [ mailto:theo.diefent...@scoop-software.de | theo.diefent...@scoop-software.de ] > wrote: Hi Marco, In general, I see three solutions here you could approach: 1. Use the StateProcessorAPI: You can run a program with the stateProcessorAPI that loads the data from JDBC and stores it into a Flink SavePoint. Afterwards, you start your streaming job from that savepoint which will load its state and within find all the data from JDBC stored already. 2. Load from master, distribute with the job: When you build up your jobgraph, you could execute the JDBC queries and put the result into some Serializable class which in turn you plug in a an operator in your stream (e.g. a map stage). The class along with all the queried data will be serialized and deserialized on the taskmanagers (Usually, I use this for configuration parameters, but it might be ok in this case as well) 3. Load from TaskManager: In your map-function, if the very first event is received, you can block processing and synchronously load the data from JDBC (So each Taskmanager performs the JDBC query itself). You then keep the data to be used for all subsequent map steps. I think, option 3 is the easiest to be implemented while option 1 might be the most elegant way in my opinion. Best regards Theo Von: "Marco Villalobos" < [ mailto:mvillalo...@kineteque.com | mvillalo...@kineteque.com ] > An: "Leonard Xu" < [ mailto:xbjt...@gmail.com | xbjt...@gmail.com ] > CC: "user" < [ mailto:user@flink.apache.org | user@flink.apache.org ] > Gesendet: Mittwoch, 5. August 2020 04:33:23 Betreff: Re: Two Queries and a Kafka Topic Hi Leonard, First, Thank you. I am currently trying to restrict my solution to Apache Flink 1.10 because its the current version supported by Amazon EMR. i am not ready to change our operational environment to solve this. Second, I am using the DataStream API. The Kafka Topic is not in a table, it is in a DataStream. The SQL queries are literally from a PostgresSQL database, and only need to be run exactly once in the lifetime of the job. I am struggling to determine where this happens. JDBCInputFormat seems to query the SQL table repetitively, and also connecting streams and aggregating into one object is very complicated. Thus, I am wondering what is the right approach. Let me restate the parameters. SQL Query One = data in PostgreSQL (200K records) that is used for business logic. SQL Query Two = data in PostgreSQL (1000 records) that is used for business logic. Kafka Topic One = unlimited data-stream that uses the data-stream api and queries above to write into multiple sinks Asci Diagram: [ SQL Query One] ----> [Aggregate to Map] Kafka ----> [Kafka Topic One] --- [Keyed Process Function (Query One Map, Query Two Map)] ---<[Multiple Sinks] [ SQL Query Two] ----> [Aggregate to Map] Maybe my graph above helps. You see, I need Query One and Query Two only ever execute once. After that the information they provide are used to correctly process the Kafka Topic. I'll take a deep further to try and understand what you said, thank you, but JDBCInputFormat seem to repetitively query the database. Maybe I need to write a RichFunction or AsyncIO function and cache the results in state after that. BQ_BEGIN On Aug 4, 2020, at 6:25 PM, Leonard Xu < [ mailto:xbjt...@gmail.com | xbjt...@gmail.com ] > wrote: Hi, Marco BQ_BEGIN If I need SQL Query One and SQL Query Two to happen just one time, Looks like you want to reuse this kafka table in one job, It’s supported to execute multiple query in one sql job in Flink 1.11. You can use `StatementSet`[1] to add SQL Query one and SQL query Two in a single SQL job[1]. Best Leonard [1] [ https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/insert.html#run-an-insert-statement | https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/insert.html#run-an-insert-statement ] BQ_BEGIN 在 2020年8月5日，04:34，Marco Villalobos < [ mailto:mvillalo...@kineteque.com | mvillalo...@kineteque.com ] > 写道： Lets say that I have: SQL Query One from data in PostgreSQL (200K records). SQL Query Two from data in PostgreSQL (1000 records). and Kafka Topic One. Let's also say that main data from this Flink job arrives in Kafka Topic One. If I need SQL Query One and SQL Query Two to happen just one time, when the job starts up, and afterwards maybe store it in Keyed State or Broadcast State, but it's not really part of the stream, then what is the best practice for supporting that in Flink The Flink job needs to stream data from Kafka Topic One, aggregate it, and perform computations that require all of the data in SQL Query One and SQL Query Two to perform its business logic. I am using Flink 1.10. I supposed to query the database before the Job I submitted, and then pass it on as parameters to a function? Or am I supposed to use JDBCInputFormat for both queries and create two streams, and somehow connect or broadcast both of them two the main stream that uses Kafka Topic One? I would appreciate guidance. Please. Thank you. Sincerely, Marco A. Villalobo BQ_END BQ_END BQ_END
