Update Postgres Using Flink


Update Postgres Using Flink



My dataset named dbData contains set of data. I want to update postgres with that data regularly where the dbData changes regularly on everyday purpose.


dbData.map(new Write())
.output(JDBCOutputFormat.buildJDBCOutputFormat()
.setDrivername(Utils.properties_fetch("drivername"))
.setDBUrl(Utils.properties_fetch("dbURL"))
.setUsername(Utils.properties_fetch("username"))
.setPassword(Utils.properties_fetch("password"))
.setQuery( Write.updatequery)
.finish());



My "Write" class looks like the following:


public class Write implements MapFunction<Tuple7<String, String, String, String, String, String, String>, Row>
{
static String updatequery ;
private static final long serialVersionUID = 1L;
public Row map( Tuple7<String, String, String, String, String, String, String> value)throws Exception
{
Row obj = new Row(7);
obj.setField(0, value.f0);
obj.setField(1, value.f1);
obj.setField(2, value.f2);
obj.setField(3, value.f3);
obj.setField(4, value.f4);
obj.setField(5, value.f5);
obj.setField(6, value.f6);
Write.updatequery=putdatainDb(obj);
return obj;
}
public String putdatainDb(Row obj)
{


String updateQuery="UPDATE dashboard SET metric_result
='"+obj.getField(2)+"' ,metric_executed_on ='"+obj.getField(5)+"' ::date
where metric_orgid ='"+obj.getField(6)+"' and date(metric_from)
='"+obj.getField(3)+"'::date and date(metric_to) ='"+obj.getField(4)+"' ::date and metric_topic ='"+obj.getField(1)+"' ;";
return updateQuery;


}
}



In set query I want the query to change every time with the new row so that I can update my database regularly. Suggest some ways to achieve this.





May I know why negative is given to this question. I was unable to find the proper solution to this question so I asked as this was one of my important requirement.
– Vijaya Seetharaman
Jul 2 at 3:29






Any help regarding this?
– Vijaya Seetharaman
Jul 2 at 8:51





People easily give down vote but wont explain why they have done it. I am sorry to say this but this is what happening. If there is a related question then send me the link. I have went across many answers and I dint find them suiting my doubt so I asked.
– Vijaya Seetharaman
Jul 2 at 11:16





What you're doing seems fine, what's the actual problem? Also, please note you're not using Flink's latest stable version.
– gcandal
Jul 2 at 16:30






In dbData you see there is .setQuery there the query is not coming properly and coming as null which i am fetching from the "Write" class. The issue I think is the query is being formed first and then the "obj" object is being created. I am using flink-1.4 version.
– Vijaya Seetharaman
Jul 3 at 4:41










By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.

-

Popular posts from this blog

api-platform.com Unable to generate an IRI for the item of type

api-platform.com Unable to generate an IRI for the item of type I'm trying to setup my first api-platform instance and have run into some presumably noob questions around IDE. What i'm trying to do is setting up an register and reset entity in order to deal with registrations and password reminders from an app. And then have the entity's passed to a custom controller to do some checks, balances and mails afterwards (haven’t gotten around to that yet). But I keep running into this error when posting to my custom entities: { "@context": "/api/contexts/Error", "@type": "hydra:Error", "hydra:title": "An error occurred", "hydra:description": "Unable to generate an IRI for the item of type "AppXXXUserBundleEntityRegister"", "trace": [ { "namespace": "", "short_class": "", "c
Read more

How to set up datasource with Spring for HikariCP?

How to set up datasource with Spring for HikariCP? Hi I'm trying to use HikariCP with Spring for connection pool. I'm using jdbcTempLate and JdbcdaoSupport. This is my spring configuration file for datasource: <bean id="dataSource" class="com.zaxxer.hikari.HikariDataSource"> <property name="dataSourceClassName" value="oracle.jdbc.driver.OracleDriver"/> <property name="dataSource.url" value="jdbc:oracle:thin:@localhost:1521:XE"/> <property name="dataSource.user" value="username"/> <property name="dataSource.password" value="password"/> </bean> But unfortunately the following error message is generating: Cannot resolve reference to bean 'dataSource' while setting bean property 'dataSource'; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'dataSource'
Read more

Display dokan vendor name on Woocommerce single product pages

Display dokan vendor name on Woocommerce single product pages With woocommerce I am using Dokan plugin and I am trying to display the vendor name, rating and vendor location on single product pages. I tried this code to display vendor name but no luck: //show store name add_action( 'woocommerce_single_product_summary', 'show_store_name', 20 ); function show_store_name() { printf( '<b>Seller Name:</b> <a href="%s">%s</a>', dokan_get_store_url( $author->ID ), $store_info['store_name'] ); } Any help is appreciated. 1 Answer 1 I don't use dokan plugin. Here is the way to get the (post) author ID and to make your code work: add_action( 'woocommerce_single_product_summary', 'display_author_name', 20 ); function display_author_name() { // Get the author ID (the vendor ID) $vendor_id = get_post_field( &#
Read more