flink sink es

[董鹏 <do...@imdada.cn>]

阿里大神，这问题无思路了。偶先。暂时不清楚会不会丢数据。从falcon对es的监控来看，es正常。2019-02-23 19:46:30.965 +0800 [Source: jd_mall_order -> Flat Map -> (Sink: Unnamed, Sink: Unnamed) (1/1)] WARN  [com.imdada.parse.PublishTimeParse] [AbstractParseModel.scala:20] - fail parse json object:{"eventType":1,"execTime":1550879841000,"schema":"plat_order","table":"jd_mall_order","tableName":"jd_mall_order","columnValues":{"update_time":"2019-02-23 07:57:21","create_time":"2019-02-23 07:57:21","jd_type":"9","deliver_company_name":"?????","jd_order_status":"0","order_source":"1","jd_express_no":"88048920204","id":"723109463","order_id":"280587990738384","jd_order_no":"88048920204","deliver_company_id":"566399"},"updatedValues":{}} org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:596) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:649) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:602) 	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:712) 	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:690) 	at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51) 	at com.imdada.dw.parse.AbstractParseModel$class.flatMap(AbstractParseModel.scala:24) 	at com.imdada.parse.PublishTimeParse.flatMap(PublishTimeParse.scala:13) 	at com.imdada.parse.PublishTimeParse.flatMap(PublishTimeParse.scala:13) 	at org.apache.flink.streaming.api.operators.StreamFlatMap.processElement(StreamFlatMap.java:50) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534) 	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:712) 	at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:690) 	at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104) 	at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111) 	at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordWithTimestamp(AbstractFetcher.java:398) 	at org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher.emitRecord(Kafka010Fetcher.java:89) 	at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:154) 	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:738) 	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:94) 	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:58) 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300) 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711) 	at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.RuntimeException: An error occurred in ElasticsearchSink. 	at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.checkErrorAndRethrow(ElasticsearchSinkBase.java:379) 	at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.invoke(ElasticsearchSinkBase.java:304) 	at org.apache.flink.streaming.api.functions.sink.SinkFunction.invoke(SinkFunction.java:52) 	at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56) 	at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579) 	... 28 common frames omitted Caused by: org.elasticsearch.client.transport.NoNodeAvailableException: None of the configured nodes were available: [{basedata04}{EQmKCsxLQZmuAHxNPS2Qwg}{qT-kaeaJRFiG3inAed7Ulg}{10.9.45.166}{10.9.45.166:9300}, {basedata01}{AHapMouBTsSz4exhmlNoLA}{7PdCQ09RRiuOP7cIHDLHLw}{10.9.25.126}{10.9.25.126:9300}, {basedata02}{DnIefl5ZSgyuGD-pjHI3FA}{4alVd-SJSQ2H6PQLdDDrKg}{10.9.23.35}{10.9.23.35:9300}, {basedata05}{5So1Bw2fS6q9FmIFLGHCtw}{t8UgC535TLSdmbJLic-MRw}{10.9.48.98}{10.9.48.98:9300}, {basedata03}{xN9nMAB5SECuKyrJvUlrsA}{bKONsxAVS7-bmDC7dGfQJw}{10.10.2.15}{10.10.2.15:9300}] 	at org.elasticsearch.client.transport.TransportClientNodesService$RetryListener.onFailure(TransportClientNodesService.java:272) 	at org.elasticsearch.action.ActionListenerResponseHandler.handleException(ActionListenerResponseHandler.java:51) 	at org.elasticsearch.transport.TransportService$ContextRestoreResponseHandler.handleException(TransportService.java:984) 	at org.elasticsearch.transport.TransportService$5.doRun(TransportService.java:552) 	at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:527) 	at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37) 	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) 	... 1 common frames omitted Caused by: org.elasticsearch.transport.SendRequestTransportException: [basedata04][10.9.45.166:9300][indices:data/write/bulk] 	at org.elasticsearch.transport.TransportService.sendRequestInternal(TransportService.java:531) 	at org.elasticsearch.transport.TransportService.sendRequest(TransportService.java:465) 	at org.elasticsearch.action.TransportActionNodeProxy.execute(TransportActionNodeProxy.java:51) 	at org.elasticsearch.client.transport.TransportProxyClient.lambda$execute$0(TransportProxyClient.java:59) 	at org.elasticsearch.client.transport.TransportClientNodesService$RetryListener.onFailure(TransportClientNodesService.java:275) 	... 8 common frames omitted Caused by: org.elasticsearch.transport.NodeNotConnectedException: [basedata04][10.9.45.166:9300] Node not connected 	at org.elasticsearch.transport.TcpTransport.nodeChannel(TcpTransport.java:491) 	at org.elasticsearch.transport.TcpTransport.sendRequest(TcpTransport.java:921) 	at org.elasticsearch.transport.TransportService.sendRequestInternal(TransportService.java:520) 	... 12 common frames omitted