使用spark/scala将JSON文件连接到数据帧中

Question

我正在尝试将两个json文件连接在一起并转换为DataFrame。下面的代码打印出两个DataFrames，一个用于test_1.json，另一个用于test_2.json，但是它们没有连接在一起。

这是一段要连接在一起的代码:有人能发现我做错了什么吗？

//CODE TO JOIN THE TWO JSON FILES TOGETHER

    users_df_2.join(users_df_1,"table").createOrReplaceTempView("new_table")

   val results = spark.sql("SELECT Scenerio, table, case WHEN gender = 'F' then 'Female' WHEN gender = 'M' then 'Male' END as gender from new_table")

package com.test.dataframe

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object dataframe {
  def main(args: Array[String]): Unit = {
    println("Apache Spark Application Started ...")

    val spark = SparkSession.builder()
      .appName("Create DataFrame from json File")
      .master("local[*]")
      .getOrCreate()

    spark.sparkContext.setLogLevel("ERROR")

    //Code Block 1 Starts Here
    val json_file_path = "/Users/Desktop/test_1.json"
    val users_df_1 = spark.read.json(json_file_path)

    users_df_1.show(10, true)
    users_df_1.printSchema()
    //Code Block 1 Ends Here

    //Code Block 2 Starts Here
    val json_multiline_file_path = "/Users/Desktop/test_2.json"

    val user_schema = StructType(Array(
      StructField("Scenerio", StringType, true),
      StructField("table", StringType, true)
    ))

    val users_df_2 = spark.read.option("multiLine", "true").json(json_multiline_file_path)

    users_df_2.show(10, false)
    users_df_2.printSchema()
    //Code Block 2 Ends Here

    //CODE TO JOIN THE TWO JSON FILES TOGETHER

    users_df_2.join(users_df_1,"table").createOrReplaceTempView("new_table")

   val results = spark.sql("SELECT Scenerio, table, case WHEN gender = 'F' then 'Female' WHEN gender = 'M' then 'Male' END as gender from new_table")

    spark.stop()
    println("Apache Spark Application Completed.")
  }
}

Answer 1

为什么你不把你的两个文件都放在'/Users/Desktop/‘里面，只要模式是一样的，spark就会读入并合并它们。

val df = spark.read.option("multiLine", "true").json('/Users/Desktop/')

或者将两个文件路径放在一个列表中：

val df = sc.read.json(file1, file2, ...)