当前位置：首页 > article >正文

Spark高级用法-数据源的读取与写入

article 2024/10/26 1:30:38

数据读取

数据写入

总结

数据读取

读文件
- read.json
- read.csv
  - csv文件有两个部分构成头部数据，也就是字段数据，行数数据
- read.orc

读数据库
- read.jdbc(jdbc连接地址,table='表名',properties={'user'=用户名,'password'=密码,'driver'='驱动信息'})

缺少连接驱动的错误

拷贝连接驱动包

# 将MySQL驱动包放入/export/server/spark/jars/目录下

cp /export/server/hive/lib/mysql-connector-java-5.1.32.jar /export/server/spark/jars/

数据库创建测试数据

create database itcast charset=utf8;

create table itcast.tb_user(
    id int,
    name varchar(20),
    age int,
    gender varchar(20)
);

insert into  itcast.tb_user values (1,'张三',20,'男');

pyspark读取数据库数据

from pyspark.sql import SparkSession
ss = SparkSession.builder.getOrCreate()

# 获取外部
df = ss.read.text("hdfs://node1:8020/data/students.txt")
df.show()

# 获取外部数据库数据 采用jdbc方式读取，只要是支持jdbc连接的的数据库都可读
# url参数1  jdbc的连接地址
# table 指定连接的表
# properties 属性参数，指定连接的账户密码及驱动信息
df2 = ss.read.jdbc(
    url='jdbc:mysql://192.168.88.100:3306/itcast',table='tb_user',
    properties={'user':'root','password':'123456','driver':'com.mysql.jdbc.Driver'}
)
df2.show()

数据写入

因为数据是在df中存储，所以使用dataframe进行数据写入
- 使用dtaframe的的write方法
写入文件有个模式,覆盖和追加两种方式，用mode参数指定
- 覆盖 overwrite
- 追加 oppend

写入文件
- write.json
- write.csv
- write.orc
写入数据库
- write.jdbc(jdbc连接地址,table='表名',properties={'user'=用户名,'password'=密码,'driver'='驱动信息'},mode='写入方式')

数据库创建表

pyspark写入数据库数据

# 数据写入
from pyspark.sql import SparkSession,Row
ss = SparkSession.builder.getOrCreate()

df = ss.createDataFrame([
    Row(id = 1,name = '张三',age = 20),
    Row(id = 2,name = '李松',age = 20),
    Row(id = 3,name = '荔枝',age = 20)
],
    schema = 'id int,name string,age int'
)

# 将df数据写入hdfs中
df.write.json('hdfs://node1:8020/data/data_json',mode='overwrite')

# 写入数据库
df.write.jdbc('jdbc:mysql://192.168.88.100:3306/itcast?characterEncoding=utf8',table='tb_stu',mode='overwrite',
              properties={'user':'root','password':'123456','driver':'com.mysql.jdbc.Driver'})

验证hdfs是否写入数据