使用Apache Spark中的JDBC连接到MS SQL Server 2008 R2[复制]

1. 在提交应用程序或启动shell时包括适用的JDBC驱动程序。例如,您可以使用 --packages :

   bin/pyspark --packages group:name:version  
   

   driver-class-path 和 jars

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

   PYSPARK_SUBMIT_ARGS JVM实例启动或使用之前的环境变量 conf/spark-defaults.conf 设定 spark.jars.packages 或 spark.jars spark.driver.extraClassPath .

2. 选择所需模式。Spark JDBC writer支持以下模式:

   • append :附加此:类的内容: DataFrame 到现有数据。
   • overwrite :覆盖现有数据。
   • ignore
   • error (默认情况):如果数据已经存在,则抛出异常。

   are not supported

   mode = ...
   

3. # You can encode credentials in URI or pass
   # separately using properties argument
   # of jdbc method or options
   
   url = "jdbc:postgresql://localhost/foobar"
   

4. (可选)创建JDBC参数字典。

   properties = {
       "user": "foo",
       "password": "bar"
   }
   

   properties / options 也可以用来设置 supported JDBC connection properties .

5. DataFrame.write.jdbc

   df.write.jdbc(url=url, table="baz", mode=mode, properties=properties)
   

   保存数据(请参见 pyspark.sql.DataFrameWriter

已知问题

• 使用包含驱动程序时,找不到合适的驱动程序 --包装 java.sql.SQLException: No suitable driver found for jdbc: ... )

  假设没有驱动程序版本不匹配来解决这个问题,您可以添加 driver 属性 . 例如:

  properties = {
      ...
      "driver": "org.postgresql.Driver"
  }
  

• 使用 df.write.format("jdbc").options(...).save() 可能导致:

  解决方案未知。

• 在Pyspark 1.3中,可以尝试直接调用Java方法:

  df._jdf.insertIntoJDBC(url, "baz", True)
  

读取数据

1. 写入数据

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())
   

:

• sparksql支持JDBC源的谓词下推,但并非所有谓词都可以下推。它也不授权限制或聚合。可能的解决方法是更换 dbtable / table 具有有效子查询的参数。示例见:

  • Does spark predicate pushdown work with JDBC?
  • More than one hour to execute pyspark.sql.DataFrame.take(4)
  • How to use SQL query to define table in dbtable?

• • 提供分区 column (必须是 IntegeType ), lowerBound upperBound , numPartitions
  • 提供互斥谓词的列表 predicates

  参见:

  • Partitioning in spark while reading from RDBMS via JDBC
  • How to optimize partitioning when migrating data from JDBC source? ,
  • How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
  • How to partition Spark RDD when importing Postgres using JDBC?

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

• Maven Repository (获得所需的坐标 --包装 compile-group:name:version 替换相应字段)或 Maven Central Repository

  • PostgreSQL
  • MySQL

根据数据库的不同,可能会存在专门的源,在某些情况下,它是首选的:

• Pivotal Greenplum-Spark Connector
• 阿帕奇凤凰城- Apache Spark Plugin
• Microsoft SQL Server- Spark connector for Azure SQL Databases and SQL Server
• 亚马逊红移- Databricks Redshift connector (当前版本仅在专有的Databricks运行时可用。 Discontinued open source version, available on GitHub

下载 mysql连接器java 驱动程序并保存在spark jar文件夹中,观察下面的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")

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

https://jaceklaskowski.gitbooks.io/mastering-apache-spark/exercises/spark-exercise-dataframe-jdbc-postgresql.html “/home/anand/.ivy2/jars”/org.postgresql\u postgresql-42.1.1.震击器“

如果你的spark版本是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()