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

时间:2024-10-17 11:20:22

目录

数据读取

数据写入

总结


数据读取

  • 读文件

    • 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是否写入数据

验证数据库是否传入数据

总结

使用read和write实现数据导入导出

读取mysql数据库的原始数据表

df = ss.read.jdbc()

在将读取到的数据导入数仓中

df.write.orc(hdfs://node1:8020/ods/tb_user