Flink：错误解析数值格式

Question

我正在尝试使用 Zeppelin 在 Flink (Scala) 中开发 K-means 模型。这是我的简单代码的一部分：

//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$$anonfun$applyOrElse.apply$mcV$sp(JobManager.scala:822)
  at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$$anonfun$applyOrElse.apply(JobManager.scala:768)
  at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$$anonfun$applyOrElse.apply(JobManager.scala:768)
  at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1(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 解析器。

Flink：错误解析数值格式

Flink: ERROR parse numeric value format

scala

apache-flink

knn

flinkml