Как преобразовать строку в поток данных Apache Flink в Scala?

Я пишу скрипт Scala для обработки файла csv в Apache Flink, используя API потока данных. Мне нужно исправить формат для некоторых столбцов, а затем привести их к нужным типам.

Мой текущий код таков:

package org.myorg.tareac

import org.apache.flink.api.scala._
import org.apache.flink.api.common.functions.{FilterFunction, FlatMapFunction, MapFunction}

object BatchJob {

  def main(args: Array[String]) {
    // set up the batch execution environment
    val env = ExecutionEnvironment.getExecutionEnvironment
    
    val inputPath = "file:////home/almu/Practicas_BigData/Tarea_Flink_B/RIA_exportacion_datos_diarios_Huelva_20140206.csv"
    
    val input = env.readCsvFile[(String, String, String, String, String, String,
                                 String, String, String, String, String, String,
                                 String, String, String, String, String, String)](inputPath, fieldDelimiter=";", ignoreFirstLine=true)


    input.print()
    
    val casted_data = input.flatMap((IDPROVINCIA: String, SPROVINCIA: String, IDESTACION: String, SESTACION: String,
                                     FECHA: String, ANIO: String, TEMPMAX: String, HORMINTEMPMAX: String, TEMPMIN: String,
                                     HORMINTEMPMIN: String, TEMPMEDIA: String, HUMEDADMAX: String, HUMEDADMIN: String,
                                     HUMEDADMEDIA: String, VELVIENTO: String, DIRVIENTO: String, RADIACION: String, PRECIPITACION: String) => {
    
                                  IDESTACION.replace("\"", "").cast(Types.Int);
                                  SESTACION.replace("\"", "");
                                  FECHA.substring(6,9).cast(Int);
                                  RADIACION.replace(",", ".").replace("", 0).cast(Double);
                                  PRECIPITACION.replace(",", ".").replace("", 0).cast(Double) 
                                 })
                                 

    // execute program
    env.execute("Flink Batch CSV Scala Processing")
  }
}

Но когда я выполняю mvn clean package, я получаю эту ошибку:

[ERROR] /home/almu/Practicas_BigData/Tarea_Flink_B/tareac/src/main/scala/batch/ProcessFileBatch.scala:54: error: value cast is not a member of String
[ERROR]                                   IDESTACION.replace("\"", "").cast(Types.Int);
[ERROR]                                                                ^
[ERROR] one error found

Как правильно сделать заброс?


scala apache-flink data-stream
person AlmuHS    schedule 22.02.2021    source источник

Ответы (2)


arrow_upward
2
arrow_downward

содержимое файла:

Джек, 12 лет, num_123, Пекин

Код:

  val input = env.readCsvFile[(String, String, String, String)](inputPath, fieldDelimiter = ",", ignoreFirstLine = false)

    input
      .map((value: (String, String, String, String)) => {
        (value._1, value._2.toInt, value._3.substring(value._3.indexOf("_") + 1).toInt)
      })
      .print()

результат:

(Jack,12,123)
person levi    schedule 22.02.2021

arrow_upward
1
arrow_downward

Замените .cast(Types.Int) на .toInt.

person Juan Miguel Boyero Corral    schedule 22.02.2021