lukecwik
commented
4 years ago You should make your ElasticsearchRestClientConfig implement Serializable.
Beam has to make your DoFns available when processing on several workers and does this in Beam Java by relying on Java serialization. So everything in your DoFns closure needs to be serializable.
On Wed, Apr 22, 2020 at 10:04 PM nithinsridhar notifications@github.com wrote:
I'm reading data from elasticsearch using rest high-level client using apache beam and I'm getting below error: Caused by: java.io.NotSerializableException: com.beam.example.config.ElasticsearchRestClientConfig
====================Elastic Search Config===================== @configuration https://github.com/configuration public class ElasticsearchRestClientConfig {
private String elasticsearchHost;
@Bean(destroyMethod = "close") public RestHighLevelClient client() {
RestHighLevelClient client = new RestHighLevelClient( RestClient.builder(new HttpHost("http://127.0.0.1:9200"))); System.out.println("SSS"); System.out.println(client); return client;}
} =====================Transformer============================== @component https://github.com/component public class ElasticsearchReadTransform extends DoFn<String,String> implements Serializable {
private ElasticsearchRestClientConfig elasticsearchRestClientConfig = new ElasticsearchRestClientConfig();
@ProcessElement public void processElement(ProcessContext pc) { String msg = pc.element(); String response = searchByProductId("ItemA26","INV360_Org_A27"); pc.output(response); }
public String searchByProductId(String productId, String orgId) { SearchRequest searchRequest = new SearchRequest(); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
QueryBuilder queryBuilder = QueryBuilders .boolQuery() .must(QueryBuilders .matchQuery("inventory_audit.PRODUCT_ID", productId) ); searchSourceBuilder.query(queryBuilder); searchRequest.source(searchSourceBuilder); SearchResponse response = null; try{ response = elasticsearchRestClientConfig.client().search(searchRequest, RequestOptions.DEFAULT); }catch (Exception e){ } return response.toString();}
}
— You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub https://github.com/GoogleCloudPlatform/DataflowJavaSDK/issues/651, or unsubscribe https://github.com/notifications/unsubscribe-auth/ACM4V3A7PIJZ6A4QDZO7UFDRN7D7PANCNFSM4MOXUSZA .
nithinsridhar
nitinsridar
I'm reading data from elasticsearch using rest high-level client using apache beam and I'm getting below error: Caused by: java.io.NotSerializableException: com.beam.example.config.ElasticsearchRestClientConfig
====================Elastic Search Config===================== @Configuration public class ElasticsearchRestClientConfig {
} =====================Transformer============================== @Component public class ElasticsearchReadTransform extends DoFn<String,String> implements Serializable {
}