I aim to create a project related to Kafka > Flink > ElasticSearch > Kibana with real time processing.

I can consume messages from Kafka in Flink but can not to connect Flink and ElasticSearch. How can I send kafka messages Flink consumed to ElasticSearch?

My python 3.8 environment includes: apache-flink=1.15.0

CodePudding user response：

You can use the Table API to create an Elasticsearch Sink table:

https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/python/table/python_table_api_connectors/#how-to-use-connectors

https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/elasticsearch/#how-to-create-an-elasticsearch-table

If you need to convert your DataStream to the Table API you can find some help in here: https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/data_stream_api/