You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by 董鹏 <do...@imdada.cn> on 2019/02/19 03:16:52 UTC
完整的异常信息
2019-02-19 10:39:51.845 +0800 [Source: order -> Flat Map -> Map -> Sink: Unnamed (1/1)] WARN [com.imdada.parse.TspOrderParse] [AbstractParseModel.scala:20] - fail parse json object:{"eventType":2,"execTime":1550543991000,"schema":"dada_order","table":"order","tableName":"order","columnValues":{"order_destination_lat":"39.892000","receiver_id":"1692182597","order_user_review_text":"","order_destination_address":"??????????-??5??1??602","area_id":"29","tips":"0.0","order_supplier_lat":"39.879460","fetch_time":"1550541421","supplier_contact_id":"3162969","order_status":"3","complaint_id":"","update_time":"1550541421","prepared_cash":"0.0","extend_id":"1692121282","deliver_fee":"10.0","pay_type":"2","id":"280498222677694","order_weight":"0.61","order_type":"1","accept_time":"1550540608","transporter_id":"4230302","order_supplier_review_text":"","cancel_time":"0","source_from":"jdDj","order_cancel_reason":"","expect_finish_time":"1550555403","create_time":"1550537403","order_receipt_url":"","order_receive_verification_code":"","allowance":"0.9","block_id":"321","finish_time":"0","order_info":"????:#3 #|????????????????","order_supplier_lng":"116.317480","expect_fetch_time":"1550538003","last_update_time":"2019-02-19 09:57:01","order_fetch_verification_code":"","order_cargo_type":"9","platform_id":"0","order_review":"","order_tool":"","supplier_id":"2425881","city_id":"2","order_value":"29.0","order_destination_lng":"116.336210"},"updatedValues":{"order_status":"4","last_update_time":"2019-02-19 10:39:51","update_time":"1550543991","finish_time":"1550543991"}} 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.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.TspOrderParse.flatMap(TspOrderParse.scala:14) at com.imdada.parse.TspOrderParse.flatMap(TspOrderParse.scala:14) 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: 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.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.StreamMap.processElement(StreamMap.java:41) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579) ... 26 common frames omitted 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) ... 32 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: [basedata03][10.10.2.15: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: [basedata03][10.10.2.15: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
------------------ Original ------------------
From: "董鹏"<do...@imdada.cn>;
Date: Tue, Feb 19, 2019 11:08 AM
To: "user-zh"<us...@flink.apache.org>;
Subject: flink sink es execption
es 5.x
flink bulk sink
现象：运行几秒钟后抛异常
es数据可以正确插入
这个异常导致cpu上升
Caused by: org.elasticsearch.transport.SendRequestTransportException: [basedata03][10.10.2.15: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: [basedata03][10.10.2.15: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
Re:完整的异常信息
Posted by 董鹏 <do...@imdada.cn>.
找到问题了：我有四条流并行写es，冲突导致。 只是异常抛的没搞明白
------------------ Original ------------------
From: "董鹏"<do...@imdada.cn>;
Date: Tue, Feb 19, 2019 11:16 AM
To: "user-zh"<us...@flink.apache.org>;
Subject: 完整的异常信息
2019-02-19 10:39:51.845 +0800 [Source: order -> Flat Map -> Map -> Sink: Unnamed (1/1)] WARN [com.imdada.parse.TspOrderParse] [AbstractParseModel.scala:20] - fail parse json object:{"eventType":2,"execTime":1550543991000,"schema":"dada_order","table":"order","tableName":"order","columnValues":{"order_destination_lat":"39.892000","receiver_id":"1692182597","order_user_review_text":"","order_destination_address":"??????????-??5??1??602","area_id":"29","tips":"0.0","order_supplier_lat":"39.879460","fetch_time":"1550541421","supplier_contact_id":"3162969","order_status":"3","complaint_id":"","update_time":"1550541421","prepared_cash":"0.0","extend_id":"1692121282","deliver_fee":"10.0","pay_type":"2","id":"280498222677694","order_weight":"0.61","order_type":"1","accept_time":"1550540608","transporter_id":"4230302","order_supplier_review_text":"","cancel_time":"0","source_from":"jdDj","order_cancel_reason":"","expect_finish_time":"1550555403","create_time":"1550537403","order_receipt_url":"","order_receive_verification_code":"","allowance":"0.9","block_id":"321","finish_time":"0","order_info":"????:#3 #|????????????????","order_supplier_lng":"116.317480","expect_fetch_time":"1550538003","last_update_time":"2019-02-19 09:57:01","order_fetch_verification_code":"","order_cargo_type":"9","platform_id":"0","order_review":"","order_tool":"","supplier_id":"2425881","city_id":"2","order_value":"29.0","order_destination_lng":"116.336210"},"updatedValues":{"order_status":"4","last_update_time":"2019-02-19 10:39:51","update_time":"1550543991","finish_time":"1550543991"}} 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.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.TspOrderParse.flatMap(TspOrderParse.scala:14) at com.imdada.parse.TspOrderParse.flatMap(TspOrderParse.scala:14) 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: 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.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.StreamMap.processElement(StreamMap.java:41) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579) ... 26 common frames omitted 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) ... 32 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: [basedata03][10.10.2.15: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: [basedata03][10.10.2.15: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
------------------ Original ------------------
From: "董鹏"<do...@imdada.cn>;
Date: Tue, Feb 19, 2019 11:08 AM
To: "user-zh"<us...@flink.apache.org>;
Subject: flink sink es execption
es 5.x
flink bulk sink
现象：运行几秒钟后抛异常
es数据可以正确插入
这个异常导致cpu上升
Caused by: org.elasticsearch.transport.SendRequestTransportException: [basedata03][10.10.2.15: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: [basedata03][10.10.2.15: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