from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.elasticsearch import Elasticsearch7SinkBuilder, ElasticsearchEmitter
import json

env = StreamExecutionEnvironment.get_execution_environment()
env.add_jars("file:///C:/flink-1.20.1/lib/flink-sql-connector-elasticsearch7-3.1.0-1.20.jar")
env.add_jars("file:///C:/flink-1.20.1/lib/flink-connector-elasticsearch-base-3.1.0-1.20.jar")

data_stream = env.from_collection([
    {"id": "1", "name": "John", "age": 25},
    {"id": "2", "name": "Jane", "age": 30}
])

# Map to JSON string
json_stream = data_stream.map(lambda x: json.dumps(x))

# Use string format for hosts - much simpler
es7_sink = Elasticsearch7SinkBuilder() \
    .set_bulk_flush_max_actions(1) \
    .set_emitter(ElasticsearchEmitter.static_index('my-index', 'id')) \
    .set_hosts(['http://10.0.0.102:9200']) \
    .set_connection_username('*****') \
    .set_connection_password('*****') \
    .build()

json_stream.sink_to(es7_sink)
env.execute("Flink Elasticsearch 7 Job")