如何使用SQLContext从pyspark笔记本执行任意sql

Question

我正在尝试一个基本的测试用例，将数据从dashDB读取到spark中，然后再次将其写回dashDB。

第1步。首先在笔记本中，我读取了数据：

sqlContext = SQLContext(sc)
dashdata = sqlContext.read.jdbc(
     url="jdbc:db2://bluemix05.bluforcloud.com:50000/BLUDB:user=****;password=****;",  
     table="GOSALES.BRANCH"
).cache()

步骤2。然后，我从dashDB创建目标表：

DROP TABLE ****.FROM_SPARK;
CREATE TABLE ****.FROM_SPARK AS (
    SELECT *
    FROM GOSALES.BRANCH
) WITH NO DATA

步骤3。最后，在笔记本中，我将数据保存到表中：

from pyspark.sql import DataFrameWriter
writer = DataFrameWriter(dashdata)
dashdata = writer.jdbc(
    url="jdbc:db2://bluemix05.bluforcloud.com:50000/BLUDB:user=****;password=****;",  
    table="****.FROM_SPARK"
)

问题：是否可以从pyspark运行步骤2中的sql？我从pyspark文档中看不出这是怎么做到的。我不想使用vanilla python来连接dashDB，因为设置库需要付出很多努力。

Answer 1

使用ibmdbpy。请参阅此简短演示。

使用as_idadataframe（），您可以将DataFrames作为表上传到dashDB中。

---

此处添加了关键步骤，因为stackoverflow不喜欢链接到答案：

步骤1:添加包含以下内容的单元格：

#!pip install --user future
#!pip install --user lazy
#!pip install --user jaydebeapi
#!pip uninstall --yes ibmdbpy
#!pip install ibmdbpy --user --no-deps
#!wget -O $HOME/.local/lib/python2.7/site-packages/ibmdbpy/db2jcc4.jar https://ibm.box.com/shared/static/lmhzyeslp1rqns04ue8dnhz2x7fb6nkc.zip 

步骤2:然后从另一个笔记本细胞

from ibmdbpy import IdaDataBase
idadb = IdaDataBase('jdbc:db2://<dashdb server name>:50000/BLUDB:user=<dashdb user>;password=<dashdb pw>') 
....

Answer 2

Yes,
You can create table in dashdb from Notebook.
Below is the code for Scala :
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.sql._
import org.apache.log4j.Logger
import org.apache.log4j.Level
import java.sql.Connection
import java.sql.DriverManager
import java.sql.SQLException
import com.ibm.db2.jcc._
import java.io._
    val jdbcClassName="com.ibm.db2.jcc.DB2Driver"
    val url="jdbc:db2://awh-yp-small02.services.dal.bluemix.net:50001/BLUDB:sslConnection=true;" // enter the hostip fromc connection settings
    val user="<username>" 
    val password="<password>" 
    Class.forName(jdbcClassName)
    val connection = DriverManager.getConnection(url, user, password)
    val stmt = connection.createStatement()
stmt.executeUpdate("CREATE TABLE COL12345(" +

                     "month VARCHAR(82))")
    stmt.close()
        connection.commit()
      connection.close()