Example PopularPlacesFromKafka fails to run
Locked
 
|
Hi,
I fail to run the PopularPlacesFromKafka example with the following exception, and I wonder what might cause this "Invalid record" error? when running within Intellij IDEA --> 07:52:23.960 [Source: Custom Source -> Map (7/8)] INFO org.apache.flink.runtime.taskmanager.Task - Source: Custom Source -> Map (7/8) (930e95aac65cbda39d9f1eaa41891253) switched from RUNNING to FAILED. java.lang.RuntimeException: Invalid record: 4010,2013003778,2013003775,START,2013-01-01 00:13:00,1970-01-01 00:00:00,-74.00074,40.7359,-73.98559,40.739063,1 at com.dataartisans.flinktraining.exercises.datastream_java.datatypes.TaxiRide.fromString(TaxiRide.java:119) ~[flink-training-exercises-0.15.1.jar:na] at com.dataartisans.flinktraining.exercises.datastream_java.utils.TaxiRideSchema.deserialize(TaxiRideSchema.java:37) ~[flink-training-exercises-0.15.1.jar:na] at com.dataartisans.flinktraining.exercises.datastream_java.utils.TaxiRideSchema.deserialize(TaxiRideSchema.java:28) ~[flink-training-exercises-0.15.1.jar:na] at org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper.deserialize(KeyedDeserializationSchemaWrapper.java:42) ~[flink-training-exercises-0.15.1.jar:na] at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:139) ~[flink-training-exercises-0.15.1.jar:na] at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:652) ~[flink-training-exercises-0.15.1.jar:na] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:86) ~[flink-streaming-java_2.11-1.4.2.jar:1.4.2] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:55) ~[flink-streaming-java_2.11-1.4.2.jar:1.4.2] at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:94) ~[flink-streaming-java_2.11-1.4.2.jar:1.4.2] at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:264) ~[flink-streaming-java_2.11-1.4.2.jar:1.4.2] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718) ~[flink-runtime_2.11-1.4.2.jar:1.4.2] at java.lang.Thread.run(Thread.java:745) [na:1.8.0_60] when deploy to and run on local cluster --> 2018-03-23 07:27:23.130 [Source: Custom Source -> Map (1/1)] INFO org.apache.flink.runtime.taskmanager.Task - Source: Custom Source -> Map (1/1) (db21c7604b94968097d4be7b8558ac08) switched from RUNNING to FAILED. java.lang.RuntimeException: Invalid record: 2264,2013002216,2013002213,START,2013-01-01 00:09:00,1970-01-01 00:00:00,-74.00402,40.742107,-73.98032,40.73522,1 at com.dataartisans.flinktraining.exercises.datastream_java.datatypes.TaxiRide.fromString(TaxiRide.java:119) ~[blob_p-bc9bbda0acf1e77543b16e023cbf2466bbe42a5d-ac2bf6479faa39e6b27477bb80f66178:na] at com.dataartisans.flinktraining.exercises.datastream_java.utils.TaxiRideSchema.deserialize(TaxiRideSchema.java:37) ~[blob_p-bc9bbda0acf1e77543b16e023cbf2466bbe42a5d-ac2bf6479faa39e6b27477bb80f66178:na] at com.dataartisans.flinktraining.exercises.datastream_java.utils.TaxiRideSchema.deserialize(TaxiRideSchema.java:28) ~[blob_p-bc9bbda0acf1e77543b16e023cbf2466bbe42a5d-ac2bf6479faa39e6b27477bb80f66178:na] at org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper.deserialize(KeyedDeserializationSchemaWrapper.java:42) ~[blob_p-bc9bbda0acf1e77543b16e023cbf2466bbe42a5d-ac2bf6479faa39e6b27477bb80f66178:na] at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:139) ~[blob_p-bc9bbda0acf1e77543b16e023cbf2466bbe42a5d-ac2bf6479faa39e6b27477bb80f66178:na] at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:652) ~[blob_p-bc9bbda0acf1e77543b16e023cbf2466bbe42a5d-ac2bf6479faa39e6b27477bb80f66178:na] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:86) ~[flink-dist_2.11-1.4.2.jar:1.4.2] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:55) ~[flink-dist_2.11-1.4.2.jar:1.4.2] at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:94) ~[flink-dist_2.11-1.4.2.jar:1.4.2] at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:264) ~[flink-dist_2.11-1.4.2.jar:1.4.2] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718) ~[flink-dist_2.11-1.4.2.jar:1.4.2] at java.lang.Thread.run(Thread.java:745) [na:1.8.0_60] I copied the PopularPlacesFromKafka.java from https://raw.githubusercontent.com/dataArtisans/flink-training-exercises/master/src/main/java/com/dataartisans/flinktraining/exercises/datastream_java/connectors/PopularPlacesFromKafka.java This is a UTF-8 formatted mail ----------------------------------------------- James C.-C.Yu +886988713275 |
|
|
Just figured out the data format in nycTaxiRides.gz doesn't match to the way TaxiRide.java interpreting the lines fed into it. Then I check the exercise training github and found the TaxiRide.java (https://github.com/dataArtisans/flink-training-exercises/tree/master/src/main/java/com/dataartisans/flinktraining/exercises/datastream_java/datatypes) was recently updated (like 11 days ago). After making some changes to TaxiRide.java, the example works like a charm. wget http://training.data-artisans.com/trainingData/nycTaxiRides.gz Following is the changes I made to TaxiRide.java locally (basically just the index to variable tokens): try {ride.rideId = Long.parseLong(tokens[0]); switch (tokens[3]) { case "START": ride.isStart = true; ride.startTime = DateTime.parse(tokens[4], timeFormatter); ride.endTime = DateTime.parse(tokens[5], timeFormatter); break; case "END": ride.isStart = false; ride.endTime = DateTime.parse(tokens[4], timeFormatter); ride.startTime = DateTime.parse(tokens[5], timeFormatter); break; default: throw new RuntimeException("Invalid record: " + line); } ride.startLon = tokens[6].length() > 0 ? Float.parseFloat(tokens[6]) : 0.0f; ride.startLat = tokens[7].length() > 0 ? Float.parseFloat(tokens[7]) : 0.0f; ride.endLon = tokens[8].length() > 0 ? Float.parseFloat(tokens[8]) : 0.0f; ride.endLat = tokens[9].length() > 0 ? Float.parseFloat(tokens[9]) : 0.0f; ride.passengerCnt = Short.parseShort(tokens[10]); ride.taxiId = Long.parseLong(tokens[1]); ride.driverId = Long.parseLong(tokens[2]); } catch (NumberFormatException nfe) { throw new RuntimeException("Invalid record: " + line, nfe); } This is a UTF-8 formatted mail ----------------------------------------------- James C.-C.Yu +886988713275 2018-03-23 8:06 GMT+08:00 James Yu <[hidden email]>:
|
|
|
Hi, When I proceed further to timePrediction exercise (http://training.data-artisans.com/exercises/timePrediction.html), I realize that the nycTaxiRides.gz's format is fine. The problem is in TaxiRide.toString(), the columns were serialized in wrong order. Hence the data persisted in Kafka has wrong format.Therefore I change TaxiRide.toString() to the following: public String toString() { StringBuilder sb = new StringBuilder(); sb.append(rideId).append(","); sb.append(isStart ? "START" : "END").append(","); sb.append(startTime.toString(timeFormatter)).append(","); sb.append(endTime.toString(timeFormatter)).append(","); sb.append(startLon).append(","); sb.append(startLat).append(","); sb.append(endLon).append(","); sb.append(endLat).append(","); sb.append(passengerCnt).append(","); sb.append(taxiId).append(","); sb.append(driverId); return sb.toString(); } This is a UTF-8 formatted mail ----------------------------------------------- James C.-C.Yu +886988713275 2018-03-23 9:59 GMT+08:00 James Yu <[hidden email]>:
|
Re: Example PopularPlacesFromKafka fails to run
|
|
Thanks for the feedback! Fabian2018-03-23 8:02 GMT+01:00 James Yu <[hidden email]>:
|
«
Return to (DEPRECATED) Apache Flink User Mailing List archive.
|
1 view|%1 views
| Free forum by Nabble | Edit this page |