新版本flink1.17-1.19写入ES,官方例子报错 java.lang.IllegalStateException: The elasticsearch emitter must be serializable.
Flink找不到javassist动态创建的类的原因可能是：
动态创建的类没有被正确加载到JVM中。请检查动态创建类的代码是否正确，以及是否使用了正确的类加载器。
动态创建的类所在的jar包没有被正确地添加到Flink的classpath中。请确保动态创建的类所在的jar包已经被添加到了Flink的classpath中。
Flink的版本与动态创建的类不兼容。请检查Flink的版本是否与动态创建的类兼容。
动态创建的类在编译时出现了错误。请检查动态创建的类的编译日志，看是否有错误信息。
这个错误是因为Elasticsearch的Emitter没有实现Serializable接口。要解决这个问题，你需要创建一个自定义的Emitter类，实现Serializable接口，并将其传递给Flink的writeTo方法。以下是一个示例：
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.xcontent.XContentType;
import java.io.IOException;
public class CustomElasticsearchEmitter {
public static void main(String[] args) throws Exception {
// 创建一个简单的数据流
DataStream<String> dataStream = ...;
// 创建一个自定义的ElasticsearchSinkFunction
class CustomElasticsearchSinkFunction implements ElasticsearchSinkFunction<String> {
private final String indexName;
public CustomElasticsearchSinkFunction(String indexName) {
this.indexName = indexName;
}
@Override
public void process(String record, RuntimeContext ctx, RequestIndexer requestIndexer) {
IndexRequest indexRequest = new IndexRequest(indexName);
indexRequest.source(record, XContentType.JSON);
requestIndexer.add(indexRequest);
}
}
// 使用自定义的Emitter将数据写入Elasticsearch
dataStream.addSink(new ElasticsearchSink<>(new CustomElasticsearchSinkFunction("your_index_name")));
}
}
在这个示例中，我们创建了一个名为CustomElasticsearchEmitter的类，其中包含一个main方法。在main方法中，我们首先创建了一个简单的数据流（这里省略了具体的创建过程）。然后，我们创建了一个名为CustomElasticsearchSinkFunction的内部类，该类实现了ElasticsearchSinkFunction接口，并重写了process方法。在process方法中，我们创建了一个IndexRequest对象，并将记录添加到该对象中。最后，我们将这个IndexRequest对象传递给RequestIndexer，以便将其发送到Elasticsearch。
在main方法的最后，我们使用addSink方法将自定义的Emitter添加到数据流中。这样，当数据流被处理时，它将使用我们的自定义Emitter将数据写入Elasticsearch。
