从0开始学习pyspark--pyspark的数据读取[第4节]

在PySpark中，读取文件型数据是一个常见的操作，Spark支持多种数据格式，如CSV、JSON、Parquet、Avro等。以下是一些常用的方法来读取不同格式的文件数据。

读取文本型数据

1. 读取CSV文件:
   • 使用spark.read.csv方法读取CSV文件，可以通过参数指定列分隔符、头部等信息。

   from pyspark.sql import SparkSession
   spark = SparkSession.builder \
       .appName("CSV Read Example") \
       .getOrCreate()
   df = spark.read.csv("path/to/your/csv/file.csv", header=True, inferSchema=True)
   

   • header=True表示文件包含头部信息。
   • inferSchema=True表示让Spark自动推断列的数据类型。
2. 读取JSON文件:
   • 使用spark.read.json方法读取JSON文件，可以是单个JSON文件或者一个包含多个JSON对象的文件。

   df = spark.read.json("path/to/your/json/file.json")
   

3. 读取Parquet文件:
   • 使用spark.read.parquet方法读取Parquet文件，这是一种列式存储格式，非常适合用于大数据处理。

   df = spark.read.parquet("path/to/your/parquet/file.parquet")
   

4. 读取Avro文件:
   • Spark没有内置的Avro支持，但是可以通过添加依赖并使用spark.read.format方法来读取Avro文件。

   df = spark.read.format("com.databricks.spark.avro").load("path/to/your/avro/file.avro")
   

   • 在使用Avro之前，需要确保已经将Avro的Spark插件添加到你的项目中。
5. 读取文本文件:
   • 使用spark.read.text方法读取文本文件，每一行都会成为DataFrame中的一行。

   df = spark.read.text("path/to/your/text/file.txt")
   

6. 读取其他格式:
   • 对于其他格式，可以使用spark.read.format方法指定格式，并使用load方法加载文件。

   df = spark.read.format("your_format").load("path/to/your/file")
   

在读取文件时，还可以指定其他选项，如分区信息、编码、压缩等。例如，如果文件存储在HDFS上，或者需要指定特定的文件系统，可以使用spark.read.format("csv").option("path", "hdfs://path/to/your/file.csv").load()。

读取hive数据

在PySpark中读取Hive数据需要确保你的Spark环境已经正确配置了Hive支持，并且你的Spark集群可以访问Hive Metastore。以下是一些基本步骤来在PySpark中读取Hive数据：

1. 确保Hive依赖:
   确保你的PySpark环境中包含了Hive依赖。如果你使用的是Apache Spark内置的Hive支持，通常这些依赖已经包含在内。如果你是在本地运行，可能需要添加Hive依赖到你的Spark环境中。
2. 配置Hive Metastore:
   你需要配置Spark来连接到Hive Metastore。这通常涉及到设置hive.metastore.uris参数，该参数指向Hive Metastore服务的URI。
3. 初始化SparkSession:
   使用SparkSession.builder来配置和初始化你的SparkSession，确保启用了Hive支持。
4. 读取Hive表:
   使用SparkSession的table方法来读取Hive表。
   以下是一个示例代码：

from pyspark.sql import SparkSession
# 初始化SparkSession，启用Hive支持
spark = SparkSession.builder \
    .appName("Hive Read Example") \
    .enableHiveSupport() \
    .getOrCreate()
# 读取Hive表
df = spark.table("your_database.your_table")
# 显示DataFrame的内容
df.show()

在这个例子中，your_database是Hive数据库的名称，your_table是你要读取的表的名称。
如果你需要指定Hive Metastore的URI，可以在SparkSession.builder中设置相关的Hive配置：

spark = SparkSession.builder \
    .appName("Hive Read Example") \
    .enableHiveSupport() \
    .config("hive.metastore.uris", "thrift://<metastore_host>:<port>") \
    .getOrCreate()

替换<metastore_host>和<port>为你的Hive Metastore服务的主机和端口。
请注意，如果你的Spark集群是在YARN上运行的，或者你有其他的集群管理器，你可能需要根据你的环境进行额外的配置。此外，确保你有足够的权限来访问Hive表和Metastore。

从HDFS读取数据

在PySpark中读取存储在HDFS（Hadoop Distributed File System）上的数据相对简单。你只需要确保你的Spark环境已经配置了与HDFS的连接，并且你的Spark应用程序有权限访问HDFS上的数据。
以下是一些基本步骤来在PySpark中读取HDFS数据：

1. 确保Hadoop依赖:
   确保你的PySpark环境中包含了Hadoop依赖。如果你是在本地运行，可能需要添加Hadoop的jar包到你的Spark环境中。
2. 配置HDFS连接:
   你需要配置Spark来连接到HDFS。这通常涉及到设置fs.defaultFS参数，该参数指向HDFS的NameNode的URI。
3. 初始化SparkSession:
   使用SparkSession.builder来配置和初始化你的SparkSession。
4. 读取HDFS上的数据:
   使用SparkSession的read方法来读取HDFS上的数据。你可以指定数据格式，如CSV、JSON、Parquet等。
   以下是一个示例代码：

from pyspark.sql import SparkSession
# 初始化SparkSession
spark = SparkSession.builder \
    .appName("HDFS Read Example") \
    .getOrCreate()
# 读取HDFS上的CSV文件
df = spark.read.csv("hdfs://<namenode_host>:<port>/<path_to_file>", header=True, inferSchema=True)
# 读取HDFS上的JSON文件
df = spark.read.json("hdfs://<namenode_host>:<port>/<path_to_file>")
# 读取HDFS上的Parquet文件
df = spark.read.parquet("hdfs://<namenode_host>:<port>/<path_to_file>")
# 显示DataFrame的内容
df.show()

在这个例子中，<namenode_host>和<port>是HDFS NameNode的主机和端口，<path_to_file>是HDFS上文件的路径。你需要根据你的HDFS集群配置替换这些值。
如果你的Spark集群已经在Hadoop环境中配置好了，并且你的Spark应用程序有权限访问HDFS，那么通常不需要额外配置就可以直接读取HDFS上的数据。如果你的Spark集群是在YARN上运行的，或者你有其他的集群管理器，你可能需要根据你的环境进行额外的配置。此外，确保你有足够的权限来访问HDFS上的数据。