项目配置项目配置appl…

项目配置项目配置application.properties日志配置log4j2.properties项目pom.xml文件

7.2

客户端工厂类CustomRestClientFactory7.4

回调函数类CustomRequestConfigCallback7.5

客户端配置类CustomHttpClientConfigCallback7.6

Es操作类CustomElasticsearchSinkFunction7.7

异常处理类CustomActionRequestFailureHandler7.8

作业主类ElasticsearchSinkStreamJobAdvancedDemo

Elasticsearch

Sink是用于将Flink数据流DataStream中的数据发送到Elasticsearch的组件。

它是Flink的一个连接器Connector用于实现将实时处理的结果或数据持续地写入Elasticsearch集群中的索引中。

数据源SourceFlink数据流的源头可以是各种数据源例如Kafka、文件系统、Socket等。

Elasticsearch

Sink通常是连接到Flink数据流的末端用于将最终处理结果或数据写入Elasticsearch。

Elasticsearch集群一个或多个Elasticsearch节点的集合用于存储和处理数据。

Elasticsearch提供了分布式的数据存储和搜索功能。

索引Index在Elasticsearch中索引是存储相关数据的地方类似于关系数据库中的表。

每个索引可以包含多个文档Document每个文档包含一个或多个字段Field。

文档Document在Elasticsearch中文档是最小的数据单元。

它们以JSON格式表示并存储在索引中。

Elasticsearch

Sink是Flink的一个数据接收器用于将数据流中的数据发送到Elasticsearch集群中的特定索引。

Sink负责将Flink数据流中的事件转换为Elasticsearch要求的格式并将其发送到指定的索引。

序列化与映射在将数据写入Elasticsearch之前通常需要对数据进行序列化和映射。

序列化是将数据从Flink的内部表示转换为Elasticsearch要求的JSON格式。

映射则是定义如何将Flink数据流中的字段映射到Elasticsearch文档中的字段。

并行度控制Elasticsearch

Sink支持并行度控制可以根据需要调整并发写入Elasticsearch的任务数量。

这有助于优化性能并避免对Elasticsearch集群造成过大的负载。

Sink是一个关键的组件用于将实时处理的结果或数据可靠地写入Elasticsearch中从而支持各种实时数据分析和搜索应用。

Elasticsearch

Flink您可以对数据进行各种转换和处理。

这可能包括数据清洗、转换、聚合、窗口操作等。

在您的

Flink

集群的地址、端口、索引名称等信息。

您还可以配置其他参数例如批量写入的大小、超时时间等。

数据发送到

Elasticsearch

Elasticsearch以提高写入的效率和性能。

序列化与映射

Flink

提供了容错机制来确保数据的可靠性和一致性。

如果在数据发送过程中发生错误例如网络故障或

Elasticsearch

会自动进行故障恢复并重新发送丢失的数据以确保数据不会丢失。

性能优化

Sink

可以通过调整批量写入的大小、并发度等参数来优化性能。

这可以减少与

Elasticsearch

来配置批量写入的大小、并发度等参数以优化写入性能。

TransportClient

RestHighLevelClient

Elasticsearch、执行查询、索引管理等操作。

序列化器Serializer

在将数据发送到

格式。

您可以根据具体的数据类型和业务需求来实现自定义的序列化器。

Elasticsearch

连接配置

需要具备容错和错误处理机制以确保数据的可靠性和一致性。

如果在数据发送过程中发生错误例如网络故障或

Elasticsearch

写入失败时可以启用退避策略以避免频繁的重试。

此时setBulkFlushBackoffDelay

设置批量写入的退避延迟时间,在发生写入失败后等待指定的延迟时间后再进行重试

设置批量写入的最大重试次数,设置在写入失败后的最大重试次数。

超过这个次数后将不再重试

connectTimeout

集群建立连接的超时时间单位为毫秒。

在指定的时间内无法建立连接将会抛出连接超时异常

socketTimeout

连接的套接字超时时间单位为毫秒。

该参数定义了在建立连接后从服务器读取数据的超时时间。

设置连接请求超时时间单位为毫秒。

该参数表示从连接池获取连接的超时时间。

如果在指定的时间内无法获得连接将会抛出连接请求超时异常。

redirectsEnabled

设置是否允许重定向。

如果设置为true则当遇到重定向响应时客户端将跟随重定向并继续请求如果设置为false重定向响应将被视为错误。

maxRedirects

启用身份验证功能。

通过设置该参数为true可以提供用户名和密码进行身份验证以连接到

Elasticsearch

设置是否允许循环重定向。

如果设置为true则允许在重定向过程中发生循环重定向如果设置为false则在检测到循环重定向时将会抛出异常。

Elasticsearch

机制。

当设置为true时在发送请求之前客户端会发送一个请求头部询问服务器是否接受请求的主体部分。

如果服务器响应允许继续发送请求主体则客户端会继续发送请求如果服务器响应拒绝继续发送请求主体则客户端会放弃该请求。

normalizeUri

dependencygroupIdorg.apache.flink/groupIdartifactIdflink-java/artifactIdversion1.14.4/version

/dependency

dependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-scala_1.12/artifactIdversion1.14.4/version

/dependency

dependencygroupIdorg.apache.flink/groupIdartifactIdflink-clients_1.12/artifactIdversion1.14.4/version

/dependency

dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-elasticsearch7_1.12/artifactIdversion1.14.4/version

/dependency06

com.alibaba.fastjson.JSONObject;

import

org.apache.flink.api.common.functions.RuntimeContext;

import

org.apache.flink.streaming.api.datastream.DataStreamSource;

import

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import

org.apache.flink.streaming.api.functions.source.SourceFunction;

import

org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;

import

org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;

import

org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;

import

org.elasticsearch.action.index.IndexRequest;

import

org.elasticsearch.client.Requests;

import

org.elasticsearch.common.xcontent.XContentType;

import

java.time.format.DateTimeFormatter;

import

ElasticsearchSinkStreamJobQuickDemo

{private

LoggerFactory.getLogger(ElasticsearchSinkStreamJobQuickDemo.class);public

static

创建elasticsearch集群的httpHost连接HttpHost

httpHost

ArrayList();httpHosts.add(httpHost);//

创建elasticsearchSinkFunction函数对象专门用于处理数据写入elasticsearchSink算子队列会自动创建索引ElasticsearchSinkFunctionJSONObject

new

ElasticsearchSinkFunctionJSONObject()

{Overridepublic

element.getString(transId);String

tradeTime

element.getString(tradeTime);String

index

tradeTime;logger.info(交易流水{},数据写入索引{}成功,

transId,

Requests.indexRequest().index(index).type(_doc).id(transId).source(element,

XContentType.JSON);indexer.add(indexRequest);}};//

构建elasticsearchSink算子BuilderElasticsearchSink.BuilderJSONObject

esSinkBuilder

ElasticsearchSink.Builder(httpHosts,

每个请求最多发送的文档数量esSinkBuilder.setBulkFlushMaxActions(1);//

每次发送请求的时间间隔esSinkBuilder.setBulkFlushInterval(1000);//构建elasticsearchSink算子ElasticsearchSinkJSONObject

sink

自定义数据源模拟生产环境交易接入,每秒下发一个json格式数据SourceFunctionJSONObject

dataSource

UUID.randomUUID().toString();//交易发生时间戳long

timeStamp

System.currentTimeMillis();//交易发生金额long

tradeAmount

Random().nextInt(1000);//交易名称String

tradeName

JSONObject();dataObj.put(transId,

tradeId);dataObj.put(timeStamp,

timeStamp);dataObj.put(tradeTime,

dateUtil(timeStamp));dataObj.put(tradeAmount,

tradeAmount);dataObj.put(tradeName,

tradeName);//模拟生产每隔1秒生成一笔交易Thread.sleep(1000);logger.info(源交易流水{},原始报文{},

tradeId,

dataObj.toJSONString());sourceContext.collect(dataObj);}}Overridepublic

void

创建运行环境StreamExecutionEnvironment

env

StreamExecutionEnvironment.getExecutionEnvironment();//

构建数据源DataStreamSourceJSONObject

dataStreamSource

数据源写入数据算子进行输出到elasticsearchdataStreamSource.addSink(sink);//

描述时间格式化工具类**

LocalDateTime.ofInstant(Instant.ofEpochSecond(timestamp),

formatter

DateTimeFormatter.ofPattern(yyyyMMdd);//

dateTimeFormat

formatter.format(dateTime);return

dateTimeFormat;}

启动上述作业后根据对应的交易流水号查询es或者查询es的索引数据但是索引数据一般是一段时间才更新

验证1检查索引数据变化

http://127.0.0.1:9200/_cat/indices?v

Elasticsearch

进阶实战主要是包括ElasticsearchSink的各种参数配置以及性能调优

7.1

es.cluster.indexPrefixflink_#内部批量处理器刷新前最大缓存的操作数

es.cluster.bulkFlushMaxActions1

es.cluster.bulkFlushMaxSizeMb10

es.cluster.bulkFlushInterval10000#是否启用批量写入的退避策略当Elasticsearch

写入失败时可以启用退避策略以避免频繁的重试。

此时setBulkFlushBackoffDelay

es.cluster.bulkFlushBackofffalse

#设置批量写入的退避延迟时间,在发生写入失败后等待指定的延迟时间后再进行重试

es.cluster.bulkFlushBackoffDelay10000

#设置批量写入的最大重试次数,设置在写入失败后的最大重试次数。

超过这个次数后将不再重试

es.cluster.bulkFlushBackoffRetries3#设置与

Elasticsearch

集群建立连接的超时时间单位为毫秒。

在指定的时间内无法建立连接将会抛出连接超时异常

#设置与

连接的套接字超时时间单位为毫秒。

该参数定义了在建立连接后从服务器读取数据的超时时间。

#设置连接请求超时时间单位为毫秒。

该参数表示从连接池获取连接的超时时间。

如果在指定的时间内无法获得连接将会抛出连接请求超时异常。

es.cluster.connectionRequestTimeout10000

设置是否允许重定向。

如果设置为true则当遇到重定向响应时客户端将跟随重定向并继续请求如果设置为false重定向响应将被视为错误。

es.cluster.redirectsEnabledfalse

es.cluster.maxRedirects3#启用身份验证功能。

通过设置该参数为true可以提供用户名和密码进行身份验证以连接到

Elasticsearch

es.cluster.au***nticationEnabledfalse

#设置是否允许循环重定向。

如果设置为true则允许在重定向过程中发生循环重定向如果设置为false则在检测到循环重定向时将会抛出异常。

es.cluster.circularRedirectsAllowedfalse

Elasticsearch

es.cluster.contentCompressionEnabledfalse

Expect:

机制。

当设置为true时在发送请求之前客户端会发送一个请求头部询问服务器是否接受请求的主体部分。

如果服务器响应允许继续发送请求主体则客户端会继续发送请求如果服务器响应拒绝继续发送请求主体则客户端会放弃该请求。

es.cluster.expectContinueEnabledfalse

es.cluster.normalizeUrifalse日志配置log4j2.properties

rootLogger.levelINFO

rootLogger.appenderRef.console.refConsoleAppender

appender.console.nameConsoleAppender

appender.console.layout.typePatternLayout

appender.console.layout.pattern%d{HH:mm:ss,SSS}

%-5p

log.fileD:\\tmprootLogger.levelINFO

rootLogger.appenderRef.console.refConsoleAppender

appender.console.nameConsoleAppender

appender.console.layout.typePatternLayout

appender.console.layout.pattern%d{HH:mm:ss,SSS}

%-5p

xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0

http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.aurora/groupIdartifactIdaurora_elasticsearch_connector/artifactIdversion1.0-SNAPSHOT/version!--属性设置--properties!--java_JDK版本--java.version1.8/java.version!--maven打包插件--maven.plugin.version3.8.1/maven.plugin.version!--编译编码UTF-8--project.build.sourceEncodingUTF-8/project.build.sourceEncoding!--输出报告编码UTF-8--project.reporting.outputEncodingUTF-8/project.reporting.outputEncoding!--json数据格式处理工具--fastjson.version1.2.75/fastjson.version!--log4j版本--log4j.version2.17.1/log4j.version!--flink版本--flink.version1.14.4/flink.version!--scala版本--scala.binary.version2.12/scala.binary.version/properties!--依赖管理--dependencies!--

fastJson工具类依赖

--dependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion${fastjson.version}/version/dependency!--

fastJson工具类依赖

--dependencygroupIdorg.apache.logging.log4j/groupIdartifactIdlog4j-slf4j-impl/artifactIdversion${log4j.version}/version/dependencydependencygroupIdorg.apache.logging.log4j/groupIdartifactIdlog4j-api/artifactIdversion${log4j.version}/version/dependencydependencygroupIdorg.apache.logging.log4j/groupIdartifactIdlog4j-core/artifactIdversion${log4j.version}/version/dependency!--

log4j日志框架依赖

--dependencygroupIdorg.apache.flink/groupIdartifactIdflink-java/artifactIdversion${flink.version}/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-scala_${scala.binary.version}/artifactIdversion${flink.version}/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-clients_${scala.binary.version}/artifactIdversion${flink.version}/version/dependency!--

Flink基础依赖

--dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-elasticsearch7_${scala.binary.version}/artifactIdversion${flink.version}/version/dependency!--

Flink

--/dependencies!--编译打包--buildfinalName${project.name}/finalName!--资源文件打包--resourcesresourcedirectorysrc/main/resources/directory/resourceresourcedirectorysrc/main/java/directoryincludesinclude**/*.xml/include/includes/resource/resourcespluginsplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-shade-plugin/artifactIdversion3.1.1/versionexecutionsexecutionphasepackage/phasegoalsgoalshade/goal/goalsconfigurationartifactSetexcludesexcludeorg.apache.flink:force-shading/excludeexcludeorg.google.code.flindbugs:jar305/excludeexcludeorg.slf4j:*/excludeexcluderorg.apache.logging.log4j:*/excluder/excludes/artifactSetfiltersfilterartifact*:*/artifactexcludesexcludeMETA-INF/*.SF/excludeexcludeMETA-INF/*.DSA/excludeexcludeMETA-INF/*.RSA/exclude/excludes/filter/filterstransformerstransformerimplementationorg.apache.maven.plugins.shade.resource.ManifestResourceTransformermainClasscom.aurora.demo,ElasticsearchSinkStreamingJobDemo/mainClass/transformer/transformers/configuration/execution/executions/plugin/plugins!--插件统一管理--pluginManagementplugins!--maven打包插件--plugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactIdversion${spring.boot.version}/versionconfigurationforktrue/forkfinalName${project.build.finalName}/finalName/configurationexecutionsexecutiongoalsgoalrepackage/goal/goals/execution/executions/plugin!--编译打包插件--pluginartifactIdmaven-compiler-plugin/artifactIdversion${maven.plugin.version}/versionconfigurationsource${java.version}/sourcetarget${java.version}/targetencodingUTF-8/encodingcompilerArgsarg-parameters/arg/compilerArgs/configuration/plugin/plugins/pluginManagement/build/project7.2

package

设置批量写入的退避延迟时间,在发生写入失败后等待指定的延迟时间后再进行重试*

*/private

bulkFlushBackoffDelay10000;/***

设置批量写入的最大重试次数,设置在写入失败后的最大重试次数。

超过这个次数后将不再重试*

*/private

集群建立连接的超时时间单位为毫秒。

在指定的时间内无法建立连接将会抛出连接超时异常*

*/private

连接的套接字超时时间单位为毫秒。

该参数定义了在建立连接后从服务器读取数据的超时时间。

*

*/private

设置连接请求超时时间单位为毫秒。

该参数表示从连接池获取连接的超时时间。

如果在指定的时间内无法获得连接将会抛出连接请求超时异常。

*

*/private

connectionRequestTimeout10000;/***

设置是否允许重定向。

如果设置为true则当遇到重定向响应时客户端将跟随重定向并继续请求如果设置为false重定向响应将被视为错误。

*

*/private

启用身份验证功能。

通过设置该参数为true可以提供用户名和密码进行身份验证以连接到

Elasticsearch

设置是否允许循环重定向。

如果设置为true则允许在重定向过程中发生循环重定向如果设置为false则在检测到循环重定向时将会抛出异常。

*

*/private

circularRedirectsAllowedfalse;/***

Elasticsearch

contentCompressionEnabledfalse;/***

Expect:

机制。

当设置为true时在发送请求之前客户端会发送一个请求头部询问服务器是否接受请求的主体部分。

*

如果服务器响应允许继续发送请求主体则客户端会继续发送请求如果服务器响应拒绝继续发送请求主体则客户端会放弃该请求。

*

*/private

expectContinueEnabledfalse;/***

*/private

setBulkFlushBackoffDelay(Integer

Integer

bulkFlushBackoffRetries;}public

void

setBulkFlushBackoffRetries(Integer

bulkFlushBackoffRetries;}public

Integer

connectionRequestTimeout;}public

void

setConnectionRequestTimeout(Integer

connectionRequestTimeout;}public

Boolean

setAu***nticationEnabled(Boolean

Boolean

circularRedirectsAllowed;}public

void

setCircularRedirectsAllowed(Boolean

circularRedirectsAllowed;}public

Boolean

contentCompressionEnabled;}public

void

setContentCompressionEnabled(Boolean

{this.contentCompressionEnabled

contentCompressionEnabled;}public

Boolean

setExpectContinueEnabled(Boolean

Boolean

ElasticsearchSinkBase.RestClientFactory

package

org.apache.commons.lang3.StringUtils;

import

org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory;

import

org.apache.http.message.BasicHeader;

import

org.elasticsearch.client.NodeSelector;

import

org.elasticsearch.client.RestClient;

import

org.elasticsearch.client.RestClientBuilder;/***

描述设置用于创建

ElasticsearchSinkBase.RestClientFactory

接口**

CustomRestClientFactory(ElasticsearchEntity

elasticsearchEntity)

elasticsearchEntity;}Overridepublic

void

configureRestClientBuilder(RestClientBuilder

restClientBuilder)

authorization};restClientBuilder.setDefaultHeaders(headers);//设置用于监听节点故障的监听器。

当节点发生故障时可以执行特定的操作restClientBuilder.setFailureListener(new

RestClient.FailureListener());//配置用于选择与之通信的节点的策略。

这涉及到

Elasticsearch

集群中多个节点的选择。

restClientBuilder.setNodeSelector(NodeSelector.ANY);//为每个请求设置路径前缀。

这可以用于将请求定向到特定的子路径。

if(StringUtils.isNoneBlank(elasticsearchEntity.getPathPrefix())){restClientBuilder.setPathPrefix(elasticsearchEntity.getPathPrefix());}//允许在创建每个请求的时候进行额外的请求配置。

restClientBuilder.setRequestConfigCallback(new

CustomRequestConfigCallback(elasticsearchEntity));//允许在创建

CloseableHttpClient

客户端配置。

restClientBuilder.setHttpClientConfigCallback(new

CustomHttpClientConfigCallback(elasticsearchEntity));//设置是否启用严格的废弃模式用于警告有关已弃用功能的使用。

restClientBuilder.setStrictDeprecationMode(false);}

7.4

回调函数类CustomRequestConfigCallback

package

org.apache.http.client.config.RequestConfig;

import

org.apache.http.impl.cookie.DefaultCookieSpec;

import

org.elasticsearch.client.RestClientBuilder;/***

描述*

RestClientBuilder.RequestConfigCallback

{private

CustomRequestConfigCallback(ElasticsearchEntity

elasticsearchEntity)

elasticsearchEntity;}Overridepublic

customizeRequestConfig(RequestConfig.Builder

custom)

集群建立连接的超时时间单位为毫秒。

在指定的时间内无法建立连接将会抛出连接超时异常custom.setConnectTimeout(elasticsearchEntity.getConnectTimeout());//

设置与

连接的套接字超时时间单位为毫秒。

该参数定义了在建立连接后从服务器读取数据的超时时间。

如果在指定的时间内没有读取到数据将会抛出超时异常。

custom.setSocketTimeout(elasticsearchEntity.getSocketTimeout());//

设置连接请求超时时间单位为毫秒。

该参数表示从连接池获取连接的超时时间。

如果在指定的时间内无法获得连接将会抛出连接请求超时异常。

custom.setConnectionRequestTimeout(elasticsearchEntity.getConnectionRequestTimeout());//

设置是否允许重定向。

如果设置为true则当遇到重定向响应时客户端将跟随重定向并继续请求如果设置为false重定向响应将被视为错误。

custom.setRedirectsEnabled(elasticsearchEntity.getRedirectsEnabled());//

设置最大重定向次数。

当允许重定向时该参数指定在遇到重定向响应时最多可以重定向的次数。

custom.setMaxRedirects(elasticsearchEntity.getMaxRedirects());//

设置是否允许循环重定向。

如果设置为true则允许在重定向过程中发生循环重定向如果设置为false则在检测到循环重定向时将会抛出异常。

custom.setCircularRedirectsAllowed(elasticsearchEntity.getCircularRedirectsAllowed());//

Elasticsearch

之间进行内容压缩以减少数据传输量。

custom.setContentCompressionEnabled(elasticsearchEntity.getContentCompressionEnabled());//

Expect:

机制。

当设置为true时在发送请求之前客户端会发送一个请求头部询问服务器是否接受请求的主体部分。

//

如果服务器响应允许继续发送请求主体则客户端会继续发送请求如果服务器响应拒绝继续发送请求主体则客户端会放弃该请求。

custom.setExpectContinueEnabled(elasticsearchEntity.getExpectContinueEnabled());//

URI以便消除多余和重复的斜杠等。

custom.setNormalizeUri(elasticsearchEntity.getNormalizeUri());//

Cookie

DefaultCookieSpec().toString());return

custom;}

客户端配置类CustomHttpClientConfigCallback

作用允许在创建

org.apache.http.client.config.RequestConfig;

import

org.apache.http.impl.cookie.DefaultCookieSpec;

import

org.apache.http.impl.nio.client.HttpAsyncClientBuilder;

import

org.elasticsearch.client.RestClientBuilder;/***

描述客户端配置*

RestClientBuilder.HttpClientConfigCallback

{private

elasticsearchEntity;CustomHttpClientConfigCallback(ElasticsearchEntity

elasticsearchEntity)

elasticsearchEntity;}Overridepublic

customizeHttpClient(HttpAsyncClientBuilder

custom

集群建立连接的超时时间单位为毫秒。

在指定的时间内无法建立连接将会抛出连接超时异常custom.setConnectTimeout(elasticsearchEntity.getConnectTimeout());//

设置与

连接的套接字超时时间单位为毫秒。

该参数定义了在建立连接后从服务器读取数据的超时时间。

如果在指定的时间内没有读取到数据将会抛出超时异常。

custom.setSocketTimeout(elasticsearchEntity.getSocketTimeout());//

设置连接请求超时时间单位为毫秒。

该参数表示从连接池获取连接的超时时间。

如果在指定的时间内无法获得连接将会抛出连接请求超时异常。

custom.setConnectionRequestTimeout(elasticsearchEntity.getConnectionRequestTimeout());//

设置是否允许重定向。

如果设置为true则当遇到重定向响应时客户端将跟随重定向并继续请求如果设置为false重定向响应将被视为错误。

custom.setRedirectsEnabled(elasticsearchEntity.getRedirectsEnabled());//

设置最大重定向次数。

当允许重定向时该参数指定在遇到重定向响应时最多可以重定向的次数。

custom.setMaxRedirects(elasticsearchEntity.getMaxRedirects());//

启用身份验证功能。

通过设置该参数为true可以提供用户名和密码进行身份验证以连接到

Elasticsearch

集群。

custom.setAu***nticationEnabled(elasticsearchEntity.getAu***nticationEnabled());//

设置是否允许循环重定向。

如果设置为true则允许在重定向过程中发生循环重定向如果设置为false则在检测到循环重定向时将会抛出异常。

custom.setCircularRedirectsAllowed(elasticsearchEntity.getCircularRedirectsAllowed());//

Elasticsearch

之间进行内容压缩以减少数据传输量。

custom.setContentCompressionEnabled(elasticsearchEntity.getContentCompressionEnabled());//

Expect:

机制。

当设置为true时在发送请求之前客户端会发送一个请求头部询问服务器是否接受请求的主体部分。

//

如果服务器响应允许继续发送请求主体则客户端会继续发送请求如果服务器响应拒绝继续发送请求主体则客户端会放弃该请求。

custom.setExpectContinueEnabled(elasticsearchEntity.getExpectContinueEnabled());//

URI以便消除多余和重复的斜杠等。

custom.setNormalizeUri(elasticsearchEntity.getNormalizeUri());//

Cookie

DefaultCookieSpec().toString());return

httpAsyncClientBuilder.setDefaultRequestConfig(custom.build());}

}7.6

Es操作类CustomElasticsearchSinkFunction

作用实时把数据写入到队列中再通过批量提交到Elasticsearch中实现数据写入

package

com.alibaba.fastjson.JSONObject;

import

org.apache.flink.api.common.functions.RuntimeContext;

import

org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;

import

org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;

import

org.elasticsearch.action.index.IndexRequest;

import

org.elasticsearch.client.Requests;

import

org.elasticsearch.common.xcontent.XContentType;

import

CustomElasticsearchSinkFunction

implements

ElasticsearchSinkFunctionJSONObject

{private

LoggerFactory.getLogger(CustomElasticsearchSinkFunction.class);private

ElasticsearchEntity

CustomElasticsearchSinkFunction(ElasticsearchEntity

elasticsearchEntity)

elasticsearchEntity;}Overridepublic

void

element.getString(transId);String

tradeTime

element.getString(tradeTime);String

index

elasticsearchEntity.getIndexPrefix()

tradeTime;logger.info(交易流水{},数据写入索引{}成功,

tradeTime,

Requests.indexRequest().index(index).type(elasticsearchEntity.getType()).id(transId).source(element,

XContentType.JSON);indexer.add(indexRequest);}

7.7

异常处理类CustomActionRequestFailureHandler

作用当sink写Elasticsearch出现异常时可以自定义操作策略

package

org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;

import

org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;

import

org.elasticsearch.action.ActionRequest;

import

org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;

import

CustomActionRequestFailureHandler

implements

LoggerFactory.getLogger(CustomActionRequestFailureHandler.class);Overridepublic

void

如果是由于线程池饱和导致的拒绝执行异常可以采取相应的处理措施logger.warn(Elasticsearch

action

这里你可以选择执行重试或者其他处理逻辑例如将数据写入到一个备用存储//

indexer.add(createAno***rRequest(action));}

else

对于其他类型的异常默认返回放弃策略logger.error(Unhandled

failure,

作业主类ElasticsearchSinkStreamJobAdvancedDemo

package

com.alibaba.fastjson.JSONObject;

import

org.apache.flink.api.java.utils.ParameterTool;

import

org.apache.flink.streaming.api.datastream.DataStreamSource;

import

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import

org.apache.flink.streaming.api.functions.source.SourceFunction;

import

org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;

import

org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;

import

java.time.format.DateTimeFormatter;

import

ElasticsearchSinkStreamJobAdvancedDemo

{private

LoggerFactory.getLogger(ElasticsearchSinkStreamJobAdvancedDemo.class);public

static

esClusterHttpHostHandler(elasticsearchEntity);//

创建esSinkFunction函数ElasticsearchSinkFunctionJSONObject

esSinkFunction

CustomElasticsearchSinkFunction(elasticsearchEntity);//

构建ElasticsearchSink算子builderElasticsearchSink.BuilderJSONObject

esSinkBuilder

ElasticsearchSink.Builder(httpHosts,

esSinkFunction);//

es参数配置esBuilderHandler(esSinkBuilder,

构建sink算子ElasticsearchSinkJSONObject

esSink

自定义数据源模拟生产环境交易接入,json格式数据SourceFunctionJSONObject

dataSource

UUID.randomUUID().toString();//交易发生时间戳long

timeStamp

System.currentTimeMillis();//交易发生金额long

tradeAmount

Random().nextInt(100);//交易名称String

tradeName

JSONObject();dataObj.put(transId,

tradeId);dataObj.put(timeStamp,

timeStamp);dataObj.put(tradeTime,

dateUtil(timeStamp));dataObj.put(tradeAmount,

tradeAmount);dataObj.put(tradeName,

tradeName);//模拟生产每隔1秒生成一笔交易Thread.sleep(1000);logger.info(交易接入,原始报文{},

dataObj.toJSONString());sourceContext.collect(dataObj);}}Overridepublic

void

创建运行环境StreamExecutionEnvironment

env

StreamExecutionEnvironment.getExecutionEnvironment();//

构建数据源DataStreamSourceJSONObject

dataStreamSource

构建sink算子dataStreamSource.addSink(esSink);//

运行作业env.execute();}

E:\\project\\aurora_dev\\aurora_flink_connector_elasticsearch\\src\\main\\resources\\application.properties;ParameterTool

paramsMap

ParameterTool.fromPropertiesFile(propertiesFilePath);ElasticsearchEntity

elasticsearchEntity

paramsMap.get(es.cluster.hosts);int

port

paramsMap.getInt(es.cluster.port);String

scheme

paramsMap.get(es.cluster.scheme);String

type

paramsMap.get(es.cluster.type);String

indexPrefix

paramsMap.get(es.cluster.indexPrefix);int

bulkFlushMaxActions

paramsMap.getInt(es.cluster.bulkFlushMaxActions);int

bulkFlushMaxSizeMb

paramsMap.getInt(es.cluster.bulkFlushMaxSizeMb);int

bulkFlushInterval

paramsMap.getInt(es.cluster.bulkFlushInterval);boolean

bulkFlushBackoff

paramsMap.getBoolean(es.cluster.bulkFlushBackoff);int

paramsMap.getInt(es.cluster.bulkFlushBackoffDelay);int

paramsMap.getInt(es.cluster.bulkFlushBackoffRetries);int

connectTimeout

paramsMap.getInt(es.cluster.connectTimeout);int

socketTimeout

paramsMap.getInt(es.cluster.socketTimeout);int

paramsMap.getInt(es.cluster.connectionRequestTimeout);boolean

redirectsEnabled

paramsMap.getBoolean(es.cluster.redirectsEnabled);int

maxRedirects

paramsMap.getInt(es.cluster.maxRedirects);boolean

paramsMap.getBoolean(es.cluster.au***nticationEnabled);boolean

paramsMap.getBoolean(es.cluster.circularRedirectsAllowed);boolean

paramsMap.getBoolean(es.cluster.contentCompressionEnabled);boolean

paramsMap.getBoolean(es.cluster.expectContinueEnabled);boolean

normalizeUri

paramsMap.getBoolean(es.cluster.normalizeUri);elasticsearchEntity.setHosts(hosts);elasticsearchEntity.setPort(port);elasticsearchEntity.setScheme(scheme);elasticsearchEntity.setType(type);elasticsearchEntity.setIndexPrefix(indexPrefix);elasticsearchEntity.setBulkFlushMaxActions(bulkFlushMaxActions);elasticsearchEntity.setBulkFlushMaxSizeMb(bulkFlushMaxSizeMb);elasticsearchEntity.setBulkFlushInterval(bulkFlushInterval);elasticsearchEntity.setBulkFlushBackoff(bulkFlushBackoff);elasticsearchEntity.setBulkFlushBackoffDelay(bulkFlushBackoffDelay);elasticsearchEntity.setBulkFlushBackoffRetries(bulkFlushBackoffRetries);elasticsearchEntity.setConnectTimeout(connectTimeout);elasticsearchEntity.setSocketTimeout(socketTimeout);elasticsearchEntity.setConnectionRequestTimeout(connectionRequestTimeout);elasticsearchEntity.setRedirectsEnabled(redirectsEnabled);elasticsearchEntity.setMaxRedirects(maxRedirects);elasticsearchEntity.setAu***nticationEnabled(au***nticationEnabled);elasticsearchEntity.setCircularRedirectsAllowed(circularRedirectsAllowed);elasticsearchEntity.setExpectContinueEnabled(expectContinueEnabled);elasticsearchEntity.setContentCompressionEnabled(contentCompressionEnabled);elasticsearchEntity.setNormalizeUri(normalizeUri);return

描述时间格式化工具类**

LocalDateTime.ofInstant(Instant.ofEpochSecond(timestamp),

formatter

DateTimeFormatter.ofPattern(yyyyMMdd);//

dateTimeFormat

formatter.format(dateTime);return

dateTimeFormat;}/***

esBuilderHandler(ElasticsearchSink.BuilderJSONObject

esSinkBuilder,

1esSinkBuilder.setBulkFlushMaxActions(elasticsearchEntity.getBulkFlushMaxActions());//

MBesSinkBuilder.setBulkFlushMaxSizeMb(elasticsearchEntity.getBulkFlushMaxSizeMb());//

解释每隔指定的时间间隔无论是否达到最大动作数或最大数据量都会触发批量写入esSinkBuilder.setBulkFlushInterval(elasticsearchEntity.getBulkFlushInterval());//

解释当

写入失败时可以启用退避策略以避免频繁的重试。

此时setBulkFlushBackoffDelay

参数生效。

esSinkBuilder.setBulkFlushBackoff(elasticsearchEntity.getBulkFlushBackoff());//

解释在发生写入失败后等待指定的延迟时间后再进行重试esSinkBuilder.setBulkFlushBackoffDelay(elasticsearchEntity.getBulkFlushBackoffDelay());//

解释设置在写入失败后的最大重试次数。

超过这个次数后将不再重试esSinkBuilder.setBulkFlushBackoffRetries(elasticsearchEntity.getBulkFlushBackoffRetries());//

ElasticsearchSinkFunction.FailureHandler

接口esSinkBuilder.setFailureHandler(new

CustomActionRequestFailureHandler());//

设置用于创建

ElasticsearchSinkBase.RestClientFactory

接口esSinkBuilder.setRestClientFactory(new

CustomRestClientFactory(elasticsearchEntity));}/***

描述:*

esClusterHttpHostHandler(ElasticsearchEntity

elasticsearchEntity)

elasticsearchEntity.getHosts().split(,);for

(String

elasticsearchEntity.getScheme()));}return

httpHosts;}