PySpark Azure SQL DB への接続、Select を実行し結果を CSV ファイルへ保存 No.39

sudo -H pip install pyodbc
pip show pyodbc

SQL DB から取得したデータを DataFrame に入れるため pandas もインストールします。

sudo apt-get update
sudo apt-get install python3-pip
sudo python3 -m pip install pandas

pyspark

（標準出力例）
Python 3.8.10 (default, Jun  2 2021, 10:49:15) 
[GCC 9.4.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
2021-10-xx 12:22:30,989 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
2021-10-xx 12:22:34,184 WARN util.Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.1.2
      /_/

Using Python version 3.8.10 (default, Jun  2 2021 10:49:15)
Spark context Web UI available at http://masternodetsu:4041
Spark context available as 'sc' (master = local[*], app id = local-xxxx).
SparkSession available as 'spark'.

3-3. SQL DB に接続し Data を select 後、結果を csv ファイルとして hdfs に保存します。

>>>#  Connect to SQL DB and select data from the table.
>>> import sys
>>> sys.path.append('/usr/local/lib/python3.8/dist-packages')
>>> import pyodbc
>>> cnxn=pyodbc.connect("Driver={ODBC Driver 17 for SQL Server};Server=tcp:xxx.database.windows.net,1433;Database=dbname;Uid=username;Pwd=password")
>>> cursor=cnxn.cursor()
>>> cursor.execute("select * from [dbo].[tableb]")
<pyodbc.Cursor object at 0x7exxxxx0>
>>> rows=cursor.fetchall()
>>> print(rows)
[('Travase', '7'), ('Ken', '8'), ('Haru', '12'), ('Hugo', '22')]

>>># Insert a new record and confirm the number of the inserted records
>>> count=cursor.execute("""insert into [dbo].[tableb] (LastName,id) values(?,?)""",'Nancy',30).rowcount
>>> cnxn.commit()
>>> print('Rows inserted: ' + str(count))
Rows inserted: 1

>>># show records again.
>>> rows=cursor.fetchall()
>>> print(rows)
[('Ken', '8'), ('Haru', '12'), ('Hugo', '22'), ('Nancy', '30')]

>>># create the dataframe using pandas library
>>> import pandas as pd
>>> query=f"select * from [dbo].[tableb]"
>>> pdf=pd.read_sql(query,cnxn)
>>> sparkdf = spark.createDataFrame(pdf)
>>> sparkdf.show()
+--------+---+                                                                  
|LastName| id|
+--------+---+
| Travase|  7|
|     Ken|  8|
|    Haru| 12|
|    Hugo| 22|
|   Nancy| 30|
+--------+---+

>>># save data into the csv file in hdfs path
>>> sparkdf.write.format("csv").save("appbox/sparkdf",header="true")
>>> spark.read.option("header","true").csv("appbox/sparkdf").show()
+--------+---+
|LastName| id|
+--------+---+
| Travase|  7|
|     Ken|  8|
|    Haru| 12|
|    Hugo| 22|
|   Nancy| 30|
+--------+---+