数据读取

• 读文件- 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