时间戳

首页 > 解决方案 > 为了避免手动文件错误，如何在 spark/scala 中编码列检查的动态数据类型

apache-spark - 为了避免手动文件错误，如何在 spark/scala 中编码列检查的动态数据类型

问题描述

我们得到了很多手动文件，我们需要在处理数据帧之前验证少数数据类型。有人可以建议我如何继续这个要求。基本上需要编写一个适用于许多文件的 spark 通用/通用程序。如果可能，请发送有关此电子邮件 ID 的更多详细信息以及 pathirammi1@gmail.com。

标签： apache-spark

解决方案

想知道您的文件是否有分隔符的记录（如 csv 文件）。如果是，您可以很好地将其作为文本文件读取，然后拆分基于记录和分隔符的记录并进行处理。

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object RDDFromCSVFile {

  def main(args:Array[String]): Unit ={

    def splitString(row:String):Array[String]={
      row.split(",")
    }

    val spark:SparkSession = SparkSession.builder()
      .master("local[1]")
      .appName("SparkByExample")
      .getOrCreate()
    val sc = spark.sparkContext

    val rdd = sc.textFile("randomfile.csv")

    val rdd2:RDD = rdd.map(row=>{
     val strArray = splitString(row)
      val field1 = strArray(0)
      val field2 = strArray(1)
      val field3 = strArray(3)
      val field4 = strArray(4)
      // DO custom code here and return to create RDD
    })

    rdd2.foreach(a=>println(a.toString))
  }

}

如果您有非结构化数据，那么您应该使用以下代码

import org.apache.spark.sql.SparkSession

object RDDFromWholeTextFile {

  def main(args:Array[String]): Unit = {

    val spark:SparkSession = SparkSession.builder()
      .master("local[1]")
      .appName("SparkByExample")
      .getOrCreate()
    val sc = spark.sparkContext

    val rdd = sc.wholeTextFiles("alice.txt")
    rdd.foreach(a=>println(a._1+"---->"+a._2))

  }
}

希望这可以帮助 ！！谢谢，纳文

推荐阅读