Flink：ERROR解析数值格式

Question

我试图在Flink（Scala）中使用Zeppelin开发K均值模型。 这是我简单的代码部分：

//Reading data 
val mapped : DataSet[Vector] = data.map {x => DenseVector (x._1,x._2) } 

//Create algorithm 
val knn = KNN() 
    .setK(3) 
    .setBlocks(10) 
    .setDistanceMetric(SquaredEuclideanDistanceMetric()) 
    .setUseQuadTree(false) 
    .setSizeHint(CrossHint.SECOND_IS_SMALL) 
... 
//Just to learn I use the same data predicting the model 
val result = knn.predict(mapped).collect() 

当我打印的资料或使用预测方法，我得到这个错误：

org.apache.flink.client.program.ProgramInvocationException: The program execution failed: Job execution failed. 
    at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:409) 
    at org.apache.flink.client.program.StandaloneClusterClient.submitJob(StandaloneClusterClient.java:95) 
    at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:382) 
    at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:369) 
    at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:344) 
    at org.apache.flink.client.RemoteExecutor.executePlanWithJars(RemoteExecutor.java:211) 
    at org.apache.flink.client.RemoteExecutor.executePlan(RemoteExecutor.java:188) 
    at org.apache.flink.api.java.RemoteEnvironment.execute(RemoteEnvironment.java:172) 
    at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:896) 
    at org.apache.flink.api.scala.ExecutionEnvironment.execute(ExecutionEnvironment.scala:637) 
    at org.apache.flink.api.scala.DataSet.collect(DataSet.scala:547) 
    ... 36 elided 
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed. 
    at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:822) 
    at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:768) 
    at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:768) 
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24) 
    at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24) 
    at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41) 
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:401) 
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) 
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253) 
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346) 
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) 
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) 
Caused by: org.apache.flink.api.common.io.ParseException: Line could not be parsed: '-6.59 -44.68' 
ParserError NUMERIC_VALUE_FORMAT_ERROR 
Expect field types: class java.lang.Double, class java.lang.Double 
in file: /home/borja/flink/kmeans/points 
    at org.apache.flink.api.common.io.GenericCsvInputFormat.parseRecord(GenericCsvInputFormat.java:407) 
    at org.apache.flink.api.java.io.CsvInputFormat.readRecord(CsvInputFormat.java:110) 
    at org.apache.flink.api.common.io.DelimitedInputFormat.nextRecord(DelimitedInputFormat.java:470) 
    at org.apache.flink.api.java.io.CsvInputFormat.nextRecord(CsvInputFormat.java:78) 
    at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:162) 
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:585) 
    at java.lang.Thread.run(Thread.java:748) 

我不知道这是否是我的错装数据或它与that之类的东西有关。

感谢您的帮助！ :)

Answer 1

您没有向我们展示您用于读取和分析数据的代码，这是错误发生的地方。但考虑到错误消息，我会冒险猜测您正在使用readCSVFile和由空格或制表符分隔的数据，并且未指定fieldDelimiter（默认为逗号）。如果是这种情况，请参阅docs了解如何配置CSV解析器。