Как добавить схему в файл из другого файла в spark Scala

Question

Я работаю в Spark и использую Scala

У меня есть два CSV-файла, один с именами столбцов, а другой с данными, как я могу объединить их оба, чтобы я мог создать результирующий файл со схемой и данными, а затем я должен применить операции к этому файлу, как groupby, cout и т. д., так как мне нужно посчитать различные значения из этих столбцов.

Так может кто-нибудь помочь здесь будет действительно полезным

Я написал приведенный ниже код, сделав два DF из обоих файлов после их прочтения, после чего я присоединился к обоим DF, используя объединение теперь, как я могу сделать первую строку в виде схемы, или любой другой способ продолжить это. Любой может предложить.

     val sparkConf = new SparkConf().setMaster("local[4]").setAppName("hbase sql")
val sc = new SparkContext(sparkConf)
val spark1 = SparkSession.builder().config(sc.getConf).getOrCreate()
    val sqlContext = spark1.sqlContext

val spark = SparkSession
  .builder
  .appName("SparkSQL")
  .master("local[*]")
  .getOrCreate()
import spark.implicits._
val lines = spark1.sparkContext.textFile("C:/Users/ayushgup/Downloads/home_data_usage_2018122723_1372672.csv").map(lines=>lines.split("""\|""")).toDF()  
 val header = spark1.sparkContext.textFile("C:/Users/ayushgup/Downloads/Header.csv").map(lin=>lin.split("""\|""")).toDF()

val file = header.unionAll (lines) .toDF ()

Answer 1

spark.sparkContext.textFile() вернет rdd и не выведет схему, даже если вы делаете .toDF() поверх этого rdd.

sc.textFile() для чтения неструктурированных текстовых файлов. Вы должны использовать

spark.read.format("csv").option("header",true").option("inferSchema","true").load("..path.to.csv")

чтобы получить схему из заголовков.

Лучше вам cat файлы вместе, создать заново CSV и читать их в HDFS

cat header.csv home_data_usage_2018122723_1372672.csv >> new_home_data_usage.csv

, а затем

hadoop fs -copyFromLocal new_home_data_usage.csv <hdfs_path>

затем используйте

spark.read.format("csv").option("header",true").option("inferSchema","true").load("..path.to.csv")