spark json转dataframe-摩杜云开发者社区

从JSON转换为DataFrame in Spark

简介

Apache Spark是一个用于大规模数据处理的开源分布式计算系统。它提供了丰富的API和工具，用于处理结构化和半结构化数据，包括JSON数据。在本文中，我们将深入研究如何使用Spark将JSON数据转换为DataFrame，并展示一些实际的代码示例。

Spark DataFrame简介

Spark DataFrame是Spark中用于处理结构化数据的主要API。它提供了类似于关系型数据库中表的概念，并支持SQL查询、数据操作和数据分析。DataFrame可以从多种数据源中创建，包括CSV、JSON、Parquet等。

Spark的DataFrame API提供了强大的功能，可以用于处理和转换复杂的数据结构，如JSON。下面我们将给出一个详细的示例，展示如何将JSON数据转换为DataFrame。

示例数据

假设我们有一个包含以下JSON数据的文件data.json：

{"name": "Alice", "age": 25, "city": "New York"}
{"name": "Bob", "age": 30, "city": "San Francisco"}
{"name": "Charlie", "age": 35, "city": "Los Angeles"}

创建SparkSession

在开始之前，我们需要创建一个SparkSession对象。SparkSession是Spark 2.0引入的新API，用于与Spark进行交互。它是Spark SQL、DataFrame和Dataset API的入口点。

使用以下代码创建一个SparkSession对象：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("JSON to DataFrame")
  .master("local")
  .getOrCreate()

读取JSON数据

接下来，我们使用.read.json()方法从文件中读取JSON数据并创建一个DataFrame。

val df = spark.read.json("path/to/data.json")

在上面的代码中，我们将文件路径传递给.read.json()方法，它将返回一个包含JSON数据的DataFrame。你可以将实际的文件路径替换为你自己的路径。

查看DataFrame

我们可以使用.show()方法查看DataFrame的内容。

df.show()

运行上述代码将输出DataFrame的前20行：

+-------+---+-------------+
|   name|age|         city|
+-------+---+-------------+
|  Alice| 25|     New York|
|    Bob| 30|San Francisco|
|Charlie| 35|  Los Angeles|
+-------+---+-------------+

DataFrame模式

DataFrame模式是DataFrame中列的名称和数据类型的描述。我们可以使用.printSchema()方法来查看DataFrame的模式。

df.printSchema()

运行上述代码将输出DataFrame的模式：

root
 |-- name: string (nullable = true)
 |-- age: integer (nullable = true)
 |-- city: string (nullable = true)

查询DataFrame

使用Spark DataFrame，我们可以执行SQL查询和数据操作。下面是一些常见的示例：

选择列

要选择DataFrame中的特定列，可以使用.select()方法。

df.select("name", "age").show()

运行上述代码将输出DataFrame中"name"和"age"列的内容：

+-------+---+
|   name|age|
+-------+---+
|  Alice| 25|
|    Bob| 30|
|Charlie| 35|
+-------+---+

添加新列

要向DataFrame中添加新列，可以使用.withColumn()方法。

val dfWithNewColumn = df.withColumn("age_plus_10", df("age") + 10)
dfWithNewColumn.show()

运行上述代码将在DataFrame中添加一个名为"age_plus_10"的新列，该列的值是"age"列值加上10：

+-------+---+-------------+------------+
|   name|age|         city|age_plus_10|
+-------+---+-------------+------------+
|  Alice| 25|     New York|          35|
|    Bob| 30|San Francisco|          40|
|Charlie| 35|  Los Angeles|          45|
+-------+---+-------------+------------+

过滤行

要过滤DataFrame中的行，可以使用.filter()方法。

val filteredDf = df.filter(df("age") > 30)
filteredDf.show()

运行上述代码将输出年龄大于30的行