spark sql可以从很多数据源中读写数据, 比较常用的是json文件和可使用jdbc协议的数据库.
访问json数据
官方文档: https://spark.apache.org/docs/latest/sql-data-sources-json.html
注意: json文件的每一行必须是一个json对象
从json加载数据
Dataset<Row> people = spark.read().json("examples/src/main/resources/people.json");
或者
Dataset<Row> people = spark.read().format("json").load("examples/src/main/resources/people.json");
写入数据到json
people.write().json("examples/src/main/resources/people.json");
或者
people.write().format("json").save("examples/src/main/resources/people.json");
基于jdbc访问数据库
官方文档: https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html
示例:
//读取数据
Dataset<Row> df = sparkSession.read()
.format("jdbc")
.option("url", "jdbc:mysql://192.168.1.22:3306/user?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai")
.option("user", "root")
.option("password", "123456")
.option("dbtable","task_info")
.load();
df.show();
//写入数据, 事先创建好表
df.write()
.format("jdbc")
.option("url", "jdbc:mysql://192.168.1.22:3306/user?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai")
.option("dbtable", "task_info2")
.option("user", "root")
.option("password", "123456")
.option("truncate","true")
.mode(SaveMode.Overwrite)
.save();
支持的选项如下:
| Property Name | Meaning |
|---|---|
| url | jdbc连接串, 例如:jdbc:postgresql://localhost/test?user=fred&password=secret |
| dbtable | 要访问的表名. 注意当在read path访问的时候, 它可能是SQL FROM 后面的有效内容(比如子查询). 不允许同时使用dbtable和query选项. |
| query | 将数据读取到spark的查询. 这个查询会被圆括号括起来, 作为FROM后面的子查询. Spark会为这个子查询设置一个别名. 例如, spark以下面这种形式向JDBC数据源发送查询SELECT <columns> FROM (<user_specified_query>) spark_gen_alias
|
| driver | 连接url的JDBC驱动的类名 |
| partitionColumn, lowerBound, upperBound | 这些选项必须全部设置或者全部不设置.另外, numPartitions也要被一起设置.它们描述了从多个worker并行读取时如何对表进行分区.partitionColumn必须是相关表中的数字、日期或时间戳列.注意，LowerBound和UpperBound只是用来决定分区的步幅，而不是用来过滤表中的行.所以表中的所有行都将被分区并返回.此选项仅适用于读取. |
| numPartitions | 表读写中可用于并行的最大分区数。这也决定了并发JDBC连接的最大数量。如果要写入的分区数超过此限制，则在写入之前通过调用coalesce（numpartitions）将其减少到此限制. |
| queryTimeout | 0意味着没有限制。在写入的时候，此选项效果取决于JDBC驱动程序如何实现APIsetQueryTimeout的，例如，h2 JDBC驱动程序检查每个查询的超时，而不是整个JDBC批处理。默认值为0 |
| fetchsize | JDBC每次往返行数, 这对默认值比较小的JDBC驱动有性能提升(比如Oracle默认是10行).这个选项仅对读取操作有效. |
| batchsize | JDBC批处理大小, 决定每次往返写入的行数.这个选项仅对写入有效, 默认值为1000. |
| isolationLevel | 事务隔离级别，对当前连接有效.它可以是NONE, READ_COMMITTED, READ_UNCOMMITTED, REPEATABLE_READ或SERIALIZABLE，对应于JDBC连接对象定义的标准事务隔离级别，默认为READ_UNCOMMITTED.此选项仅适用于写入. |
| sessionInitStatement | 会话初始化语句, 在会话打开时, 执行读取数据之前, 可以执行自定义的SQL语句(或者PL/SQL语句块). |
| truncate | 这是一个与JDBC写入相关的选项。启用SaveMode.Overwrite时，此选项会导致spark清空现有表，而不是删除并重新创建它。这样可以更有效率，并防止删除表元数据（如索引）。但是，在某些情况下，它将不起作用，例如当新数据具有不同的模式时。它默认为false。此选项仅适用于写入操作。 |
| cascadeTruncate | 这是一个与JDBC写入相关的选项。如果jdbc数据库（PostgreSQL和Oracle）启用并支持，则此选项允许执行truncate table t cascade（对于postgresql，执行truncate table only t cascade以防止无意中清空子表）。这将影响其他表，因此应小心使用。此选项仅适用于写入操作。它默认为所讨论的JDBC数据库的默认级联清空行为，在每个JDBCDialect中的isCascadeTruncate中指定。 |
| createTableOptions | 这是一个与JDBC写入相关的选项。如果指定，则此选项允许在创建表时设置特定于数据库的表和分区选项（例如，create table t（name string）engine=innodb）.此选项仅适用于写入操作. |
| createTableColumnTypes | 创建表时要使用的数据库列数据类型，而不是默认值。数据类型信息的格式应与创建表列语法相同（例如：“name char（64），comments varchar（1024）”）。指定的类型应为有效的Spark SQL数据类型。此选项仅适用于写入。 |
| customSchema | 用于从JDBC连接读取数据的自定义模式。例如，id DECIMAL(38, 0), name STRING。还可以指定部分字段，其他字段使用默认类型映射。例如，id DECIMAL(38, 0)。列名应该与JDBC表的相应列名相同。用户可以指定Spark SQL的相应数据类型，而不是使用默认值。此选项仅适用于读取操作。 |
| pushDownPredicate | 启用或禁用谓词下推到JDBC数据源的选项。默认值为true，在这种情况下，spark将尽可能将过滤器向下推送到JDBC数据源。否则，如果设置为false，则不会将任何过滤器向下推送到JDBC数据源，因此所有过滤器都将由spark处理。当SPARK比JDBC数据源更快地执行谓词筛选时，谓词下推通常被关闭。 |
