如何使用JDBC源在(Py)星火库中读写数据？

Question

这个问题的目的是记录：

• 在PySpark中使用JDBC连接读写数据所需的步骤
• JDBC源可能存在的问题及解决方案

这些方法只要做一些小的改动，就可以与其他受支持的语言一起工作，包括Scala和R。

Answer 1

写入数据

1. 在提交应用程序或启动shell时包括适用的JDBC驱动程序。例如，您可以使用--packages： bin/pyspark -软件包组:名称:版本

或者将driver-class-path和jars结合起来

    bin/pyspark --driver-class-path $PATH_TO_DRIVER_JAR --jars $PATH_TO_DRIVER_JAR

还可以在JVM实例启动之前使用PYSPARK_SUBMIT_ARGS环境变量来设置这些属性，或者使用conf/spark-defaults.conf设置spark.jars.packages或spark.jars / spark.driver.extraClassPath。

1. 选择所需的模式。Spark编写器支持以下模式：

- `append`: Append contents of this :class:`DataFrame` to existing data.
- `overwrite`: Overwrite existing data.
- `ignore`: Silently ignore this operation if data already exists.
- `error` (default case): Throw an exception if data already exists.

鞋面或其他细粒度修改不支持

模式= ..。

1. 准备JDBC，例如:您可以使用jdbc方法或选项url = "jdbc:postgresql://localhost/foobar“的属性参数#对URI中的凭据进行编码或单独传递#
2. (可选)创建JDBC参数字典。 属性={“用户”："foo"，“密码”："bar“} properties / options也可以用来设置支持的JDBC连接属性。
3. 使用DataFrame.write.jdbc df.write.jdbc(url=url，table="baz"，mode=mode，properties=properties)

若要保存数据(有关详细信息，请参阅pyspark.sql.DataFrameWriter )。

已知问题

• 当使用--packages (java.sql.SQLException: No suitable driver found for jdbc: ...)包含驱动程序时，无法找到合适的驱动程序。 假设没有驱动程序版本错配来解决这个问题，您可以将driver类添加到properties中。例如： 属性={ ..。“驱动”："org.postgresql.Driver“}
• 使用df.write.format("jdbc").options(...).save()可能会导致： org.apache.spark.sql.execution.datasources.jdbc.DefaultSource不允许创建表作为选择。

解未知。

• 在Pyspark 1.3中，您可以尝试直接调用Java方法： df._jdf.insertIntoJDBC(url，"baz"，True)

读取数据

1. 按照编写数据的步骤1-4执行
2. 使用sqlContext.read.jdbc： sqlContext.read.jdbc(url=url，table="baz"，properties=properties)

或sqlContext.read.format("jdbc")

    (sqlContext.read.format("jdbc")
        .options(url=url, dbtable="baz", **properties)
        .load())

已知的问题和

• 找不到合适的驱动程序--参见:写入数据
• Spark支持JDBC源的谓词下推，尽管并非所有谓词都可以向下推。它也不委托限制或聚合。可能的解决办法是用有效的子查询替换dbtable / table参数。例如，见：

- [Does spark predicate pushdown work with JDBC?](https://stackoverflow.com/q/32573991/1560062)
- [More than one hour to execute pyspark.sql.DataFrame.take(4)](https://stackoverflow.com/q/35869884/1560062)
- [How to use SQL query to define table in dbtable?](https://stackoverflow.com/q/32628717/6910411)

• 默认情况下，JDBC数据源使用单个执行器线程顺序加载数据。为了确保分布式数据加载，您可以：

- Provide partitioning `column` (must be `IntegerType`), `lowerBound`, `upperBound`, `numPartitions`.
- Provide a list of mutually exclusive predicates `predicates`, one for each desired partition.

请参见：

- [Partitioning in spark while reading from RDBMS via JDBC](https://stackoverflow.com/q/43150694/6910411),
- [How to optimize partitioning when migrating data from JDBC source?](https://stackoverflow.com/q/52603131/6910411),
- [How to improve performance for slow Spark jobs using DataFrame and JDBC connection?](https://stackoverflow.com/q/32188295/6910411)
- [How to partition Spark RDD when importing Postgres using JDBC?](https://stackoverflow.com/q/39597971/6910411)

• 在分布式模式下(使用分区列或谓词)，每个执行器在自己的事务中操作。如果同时修改源数据库，则无法保证最终视图是一致的。

在哪里找到合适的司机：

• Maven储存库 (获取--packages所需的坐标，选择所需的版本，并从表格compile-group:name:version中的Gradle选项卡复制数据，以替换各自的字段)或Maven中央储存库

- [PostgreSQL](http://mvnrepository.com/artifact/org.postgresql/postgresql)
- [MySQL](http://mvnrepository.com/artifact/mysql/mysql-connector-java)

其他选项

根据数据库的特定来源，可能存在并且在某些情况下是首选的：

• 格林梅- 枢轴格林梅-火花连接器
• 阿帕奇凤凰- Apache火花插件
• Microsoft Azure SQL数据库和Server的火花连接器
• Amazon Databricks红移连接器 (当前版本仅在专有数据库运行时可用。已停用的开放源码版本，可在GitHub上获得)。

Answer 2

下载mysql-connector-java驱动程序并保存在spark文件夹中，在这里观察下面的python代码将数据写入" acotr1 "，我们必须在mysql数据库中创建acotr1表结构。

    spark = SparkSession.builder.appName("prasadad").master('local').config('spark.driver.extraClassPath','D:\spark-2.1.0-bin-hadoop2.7\jars\mysql-connector-java-5.1.41-bin.jar').getOrCreate()

    sc = spark.sparkContext

    from pyspark.sql import SQLContext

    sqlContext = SQLContext(sc)

    df = sqlContext.read.format("jdbc").options(url="jdbc:mysql://localhost:3306/sakila",driver="com.mysql.jdbc.Driver",dbtable="actor",user="root",password="****").load()

    mysql_url="jdbc:mysql://localhost:3306/sakila?user=root&password=****"

    df.write.jdbc(mysql_url,table="actor1",mode="append")

Answer 3

请参考此链接下载用于postgres的jdbc，并按照步骤下载jar文件

https://jaceklaskowski.gitbooks.io/mastering-apache-spark/exercises/spark-exercise-dataframe-jdbc-postgresql.html jar文件将在路径中下载，如下所示。"/home/anand/.ivy2/jars/org.postgresql_postgresql-42.1.1.jar“

如果您的火花版本是2

from pyspark.sql import SparkSession

spark = SparkSession.builder
        .appName("sparkanalysis")
        .config("spark.driver.extraClassPath",
         "/home/anand/.ivy2/jars/org.postgresql_postgresql42.1.1.jar")
        .getOrCreate()

//for localhost database//

pgDF = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql:postgres") \
.option("dbtable", "public.user_emp_tab") \
.option("user", "postgres") \
.option("password", "Jonsnow@100") \
.load()


print(pgDF)

pgDF.filter(pgDF["user_id"]>5).show()

将文件保存为python并运行"python respectivefilename.py“