CODEWITHSUNDEEP
scalaapache-flinkdata-stream
AlmuHS
AlmuHS

Reputation: 347

How to cast a string in Apache Flink Datastream in Scala?

I'm writting a Scala script for processing a csv file in Apache Flink, using Datastream API. I need to fix the format to some columns, and then cast them to proper types.

My current code is this:

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")
  }
}

But, when I execute mvn clean package, I get this error:

[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

How can I do the cast correctly?

Upvotes: 1

Views: 732

Answers (1)

liliwei
liliwei

Reputation: 344

file content：

Jack,12,num_123,beijing

Code:

  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()

result:

(Jack,12,123)

Upvotes: 2

Related Questions