Spark SQL 提供了多种灵活的数据读取方式,支持多种数据源格式。以下是核心读取方法及示

linux环境下,按照如下方式进行

# 进入目录
cd /opt/installs/anaconda3/lib/python3.8/site-packages/pyspark/jars

# 上传jar包:mysql-connector-java-5.1.32.jar

例:

1. 从文件读取

(1) CSV 文件
df = spark.read \
    .format("csv") \
    .option("header", "true") \  # 包含标题行
    .option("inferSchema", "true") \  # 自动推断数据类型
    .load("/path/to/file.csv")

(2) JSON 文件
df = spark.read \
    .format("json") \
    .load("/path/to/file.json")

(3) Parquet 文件 (列式存储)
df = spark.read \
    .format("parquet") \
    .load("/path/to/file.parquet")

(4) ORC 文件
df = spark.read \
    .format("orc") \
    .load("/path/to/file.orc")


2. 从 JDBC 数据库读取

df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://host:port/db") \
    .option("dbtable", "table_name") \
    .option("user", "username") \
    .option("password", "password") \
    .load()


3. 从 Hive 表读取

df = spark.sql("SELECT * FROM hive_database.table_name")


4. 通用读取方法

支持通过格式名直接调用:

# 等效方法
df = spark.read.csv("/path/to/file.csv", header=True)
df = spark.read.json("/path/to/file.json")
df = spark.read.parquet("/path/to/file.parquet")


5. 高级配置选项

参数说明示例值
sep分隔符 (CSV)";"
encoding文件编码"UTF-8"
mergeSchema合并多个 Parquet 文件的 Schema"true"
queryJDBC 自定义查询语句"(SELECT col1 FROM t)"

6. 示例:读取带特殊配置的 CSV

df = spark.read \
    .option("sep", "|") \  # 管道分隔符
    .option("nullValue", "N/A") \  # 自定义空值标识
    .option("dateFormat", "yyyy-MM-dd") \  # 日期格式
    .csv("/data/sales_records.csv")

最佳实践

  • 优先使用 Parquet/ORC 等列式格式提升性能
  • 大数据集避免 inferSchema,显式定义 Schema
  • JDBC 连接添加 fetchsize 优化吞吐量
Logo

码道开发者社区,聚焦华为云码道 CodeArts 代码智能体,沉淀 Agent、Skill、鸿蒙开发实战内容,供开发者查阅资料、交流技术、分享工程实践

更多推荐