scala 阿萨中synapsesql的PySpark等价物

r1zhe5dt 于 2023-08-05 发布在 Scala

关注(0)|答案(2)|浏览(154)

在使用Azure Synapse Analytics（阿萨）的Spark notebook时，我可以使用Scala将CSV文件保存为专用SQL池中的表，只需两个简单的语句：

%%spark
// [1] - https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html
// [2] - https://learn.microsoft.com/en-us/azure/storage/blobs/data-lake-storage-introduction-abfs-uri
val testDF = spark.read.format("csv").option("header", "true").load(
    "abfss://earytestfs@earytestsa.dfs.core.windows.net/TIOBE-azure-backup.csv"
);
// [3] - https://www.aizoo.info/post/dropping-a-sql-table-in-your-synapse-spark-notebooks
// [4] - https://stackoverflow.com/questions/67907984/write-data-to-sql-dw-from-apache-spark-in-azure-synapse
testDF.write.mode("overwrite").synapsesql("eary_dedicated_test_sql_pool.dbo.TIOBE_test");

字符串
不幸的是，上面的[3]似乎暗示PySpark不存在synapsesql函数。自从达伦上次更新他的帖子以来，这种情况有没有改变？

注意：我不想为Apache Spark配置Azure Synapse Dedicated SQL Pool Connector。我的专用SQL池与我的Spark Pool在同一个工作空间中，所以我不觉得我应该为配置一堆身份验证选项而烦恼：*

https://learn.microsoft.com/en-us/azure/synapse-analytics/spark/synapse-spark-sql-pool-import-export?tabs=scala%2Cscala1%2Cscala2%2Cscala3%2Cscala4%2Cscala5

EDIT：下面的pyspark代码给我
“属性错误：'DataFrameWriter'对象没有属性'synapsesql'“

7号线

%%pyspark
df = spark.read.load('abfss://earytestfs@earytestsa.dfs.core.windows.net/TIOBE-azure-backup.csv', format='csv'
## If header exists uncomment line below
, header=True
)
# [5] - https://stackoverflow.com/questions/69720753/write-dataframe-to-sql-dedicated-database-using-synapse-analytics
df.write.mode("overwrite").synapsesql("eary_dedicated_test_sql_pool.dbo.TIOBE_test")

型

scala

来源：https://stackoverflow.com/questions/76813133/pyspark-equivalent-of-synapsesql-in-asa

2条答案

按热度按时间

luaexgnf1#

Python对synapsesql的支持已经有一年了。只需按照docs添加导入即可：

# Add required imports
import com.microsoft.spark.sqlanalytics
from com.microsoft.spark.sqlanalytics.Constants import Constants
from pyspark.sql.functions import col
# Get the table with synapsesql method and expose as temp view
df = spark.read.synapsesql("sandpit_ded.dbo.nation")
df.createOrReplaceTempView("vw_nation")

字符串
Python Notebook using synapsesql method

赞(0）回复(0）举报 2023-08-05

7y4bm7vi2#

下面的代码将帮助您将CSV文件保存为专用SQL池中的表：

from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("Write CSV to Dedicated SQL Pool") \
.getOrCreate()
server = "azuresynapseaug02.sql.azuresynapse.net"
database = "dedpoolaug02"
user = "sqladminuser"
password = "Welcome@1"
jdbcUrl = f"jdbc:sqlserver://{server};database={database};user={user};password={password}"
tableName = "dbo.Employees_nex_gen"
csv_file_path = "abfss://rawcontainerf1@synapsestorageaug02.dfs.core.windows.net/empcsv.csv"
df = spark.read.csv(csv_file_path, header=True, inferSchema=True)
mapped_df = df.selectExpr("emp_id AS emp_id", "emp_name AS emp_name", "emp_salary AS emp_salary", "emp_department AS emp_department")
mapped_df.write \
.format("jdbc") \
.option("url", jdbcUrl) \
.option("dbtable", tableName) \
.option("createTableColumnTypes", "emp_id INT, emp_name VARCHAR(50), emp_salary INT, emp_department VARCHAR(100)") \
.option("truncate", "true") \
.option("numPartitions", "4") \
.mode("overwrite") \
.save()

字符串

定义SQL池连接详情：

server = "xxxxxxxxxxxxxxxxxxxxxx"
database = "xxxxxxxxxxxxxx"
user = "xxxxxxxxxxxxxxxxxxx"
password = "xxxxxxxxxxx"

型

该代码设置Azure Synapse Analytics专用SQL池的连接详细信息。
server变量保存服务器名称，database保存数据库名称，user和password保存身份验证和连接到SQL池的凭据。将DataFrame写入SQL池
**将DataFrame写入SQL池：**使用write方法将mapped_df DataFrame写入Azure Synapse Analytics专用SQL池。格式被指定为“jdbc”以指示写入JDBC数据源。

mapped_df.write \
    .format("jdbc") \
    .option("url", jdbcUrl) \
    .option("dbtable", tableName) \
    .option("createTableColumnTypes", "emp_id INT, emp_name VARCHAR(50), emp_salary INT, emp_department VARCHAR(100)") \
    .option("truncate", "true") \
    .option("numPartitions", "4") \
    .mode("overwrite") \
    .save()

型

输出：

展开查看全部

赞(0）回复(0）举报 2023-08-05

我来回答

scala 阿萨中synapsesql的PySpark等价物

2条答案

相关问题

热门标签

最新问答