ตัวเชื่อมต่อ Spark สําหรับฐานข้อมูล SQL

Spark connector สําหรับฐานข้อมูล SQL เป็นไลบรารีประสิทธิภาพสูงที่ช่วยให้คุณอ่านและเขียนข้อมูลไปยัง SQL Server, Azure SQL และฐานข้อมูล SQL ใน Fabric ได้ ตัวเชื่อมต่อมีความสามารถต่อไปนี้:

• ใช้ Spark เพื่อรันการเขียนและอ่านขนาดใหญ่บน Azure SQL Database, Azure SQL Managed Instance, SQL Server บน Azure VM และฐานข้อมูล SQL ใน Fabric
• เมื่อคุณใช้ตารางหรือมุมมอง ตัวเชื่อมต่อจะสนับสนุนแบบจําลองความปลอดภัยที่ตั้งค่าไว้ที่ระดับกลไกจัดการ SQL แบบจําลองเหล่านี้รวมถึงการรักษาความปลอดภัยระดับออบเจ็กต์ (OLS) การรักษาความปลอดภัยระดับแถว (RLS) และการรักษาความปลอดภัยระดับคอลัมน์ (CLS)

ตัวเชื่อมต่อได้รับการติดตั้งไว้ล่วงหน้าในรันไทม์ Fabric ดังนั้นคุณจึงไม่จําเป็นต้องติดตั้งแยกต่างหาก

การรับรองความถูกต้อง

การตรวจสอบสิทธิ์ Microsoft Entra ถูกผสานรวมกับ Fabric

• เมื่อคุณลงชื่อเข้าใช้พื้นที่ทํางาน Fabric ข้อมูลประจําตัวของคุณจะถูกส่งไปยังกลไกจัดการ SQL โดยอัตโนมัติสําหรับการรับรองความถูกต้องและการอนุญาต
• จําเป็นต้องเปิดใช้งานและกําหนดค่า Microsoft Entra ID บนกลไกจัดการฐานข้อมูล SQL ของคุณ
• ไม่จําเป็นต้องมีการกําหนดค่าเพิ่มเติมในรหัส Spark ของคุณหากมีการตั้งค่า Microsoft Entra ID ข้อมูลประจําตัวจะถูกแมปโดยอัตโนมัติ

คุณยังสามารถใช้วิธีการรับรองความถูกต้อง SQL (โดยการระบุชื่อผู้ใช้และรหัสผ่าน SQL) หรือบริการหลัก (โดยการให้โทเค็นการเข้าถึง Azure สําหรับการรับรองความถูกต้องตามแอป)

การอนุญาต

เมื่อต้องการใช้ตัวเชื่อมต่อ Spark ข้อมูลประจําตัวของคุณ ไม่ว่าจะเป็นผู้ใช้หรือแอป ต้องมีสิทธิ์ฐานข้อมูลที่จําเป็นสําหรับกลไกจัดการ SQL เป้าหมาย สิทธิ์เหล่านี้จําเป็นในการอ่านหรือเขียนไปยังตารางและมุมมอง

สําหรับฐานข้อมูล Azure SQL, อินสแตนซ์ที่มีการจัดการ Azure SQL และ SQL Server บน Azure VM:

• ข้อมูลประจําตัวที่เรียกใช้การดําเนินการโดยทั่วไปต้องการ db_datawriter และ db_datareader สิทธิ์ และเป็นตัวเลือก db_owner สําหรับการควบคุมทั้งหมด

สําหรับฐานข้อมูล SQL ใน Fabric:

• โดยทั่วไปข้อมูลประจําตัวต้องการ db_datawriter และ db_datareader สิทธิ์ และเลือก db_ownerได้
• ตัวตนยังต้องมีสิทธิ์อ่านอย่างน้อยบนฐานข้อมูล SQL ใน Fabric ในระดับรายการ

Note

หากคุณใช้บริการหลัก จะสามารถทํางานเป็นแอป (ไม่มีบริบทของผู้ใช้) หรือเป็นผู้ใช้ได้หากเปิดใช้งานการเลียนแบบผู้ใช้ บริการหลักต้องมีสิทธิ์ฐานข้อมูลที่จําเป็นสําหรับการดําเนินการที่คุณต้องการดําเนินการ

ตัวอย่างการใช้งานและโค้ด

ในส่วนนี้ เราจะจัดเตรียมตัวอย่างโค้ดเพื่อสาธิตวิธีใช้ตัวเชื่อมต่อ Spark สําหรับฐานข้อมูล SQL อย่างมีประสิทธิภาพ ตัวอย่างเหล่านี้ครอบคลุมสถานการณ์ต่างๆ รวมถึงการอ่านและการเขียนไปยังตาราง SQL และการกําหนดค่าตัวเลือกตัวเชื่อมต่อ

Note

ก่อนการเขียนแบบกลุ่ม ข้อมูล Spark ที่เข้ามาทั้งหมดต้องตรงกับชนิดข้อมูล SQL เป้าหมาย เมื่อคุณเขียนทับหรือสร้างตาราง ตัวเชื่อมต่อจะแมป Spark TimestampType และTimestampNTZTypeค่าต่าง ๆ เป็น SQL datetime2 แทนที่จะเป็นdatetime ประเภท timestamp ของ Spark รองรับความแม่นยําสูงสุดหกหลักในเศษส่วนวินาที แต่ SQL datetime รองรับสามหลัก ซึ่งอาจทําให้ไม่ตรงกันได้

ตัวเลือกที่รองรับ

ตัวเลือกขั้นต่ําที่จําเป็นคือ url as "jdbc:sqlserver://<server>:<port>;database=<database>;" หรือ set spark.mssql.connector.default.url.

• เมื่อมีการ url ให้:

  • ใช้ url เป็นอันดับแรกเสมอ
  • หาก spark.mssql.connector.default.url ไม่ได้ตั้งค่า ตัวเชื่อมต่อจะตั้งค่าและนํากลับมาใช้ใหม่เพื่อการใช้งานในอนาคต

• เมื่อไม่ได้ระบุไว้ url :

  • ถ้า spark.mssql.connector.default.url ตั้งค่าไว้ ตัวเชื่อมต่อจะใช้ค่าจากการกําหนดค่าประกายไฟ
  • หาก spark.mssql.connector.default.url ไม่ได้ตั้งค่า จะมีข้อผิดพลาดเนื่องจากไม่มีรายละเอียดที่จําเป็น

ตัวเชื่อมต่อนี้สนับสนุนตัวเลือกที่กําหนดไว้ที่นี่: ตัวเลือก JDBC ของแหล่งข้อมูล SQL

ตัวเชื่อมต่อยังรองรับตัวเลือกต่อไปนี้:

ตัวเลือก	ค่าพื้นฐาน	คำอธิบาย
reliabilityLevel	"BEST_EFFORT"	ควบคุมความน่าเชื่อถือของการทํางานของเม็ดมีด ค่าที่เป็นไปได้: BEST_EFFORT (ค่าเริ่มต้น เร็วที่สุด อาจส่งผลให้เกิดแถวที่ซ้ํากันหากตัวดําเนินการรีสตาร์ท) NO_DUPLICATES (ช้ากว่า ทําให้แน่ใจว่าไม่มีการแทรกแถวที่ซ้ํากันแม้ว่าตัวดําเนินการจะรีสตาร์ทก็ตาม) เลือกตามความคลาดเคลื่อนของคุณสําหรับรายการที่ซ้ํากันและความต้องการด้านประสิทธิภาพ
isolationLevel	"READ_COMMITTED"	ตั้งค่าระดับการแยกธุรกรรมสําหรับการดําเนินการ SQL ค่าที่เป็นไปได้: READ_COMMITTED (ค่าเริ่มต้น ป้องกันการอ่านข้อมูลที่ไม่ได้ผูกมัด), READ_UNCOMMITTED, , REPEATABLE_READSNAPSHOT, . SERIALIZABLE ระดับการแยกที่สูงขึ้นสามารถลดการทํางานพร้อมกันได้ แต่ปรับปรุงความสอดคล้องของข้อมูล
tableLock	"เท็จ"	ควบคุมว่าจะใช้คําใบ้การล็อกระดับตาราง TABLOCK ของ SQL Server ในระหว่างการดําเนินการแทรกหรือไม่ ค่าที่เป็นไปได้: true (เปิดใช้งาน TABLOCK ซึ่งสามารถปรับปรุงประสิทธิภาพการเขียนจํานวนมากได้) false (ค่าเริ่มต้น ไม่ใช้ TABLOCK) การตั้งค่าเป็น true อาจเพิ่มปริมาณงานสําหรับเม็ดมีดขนาดใหญ่ แต่สามารถลดการทํางานพร้อมกันสําหรับการดําเนินการอื่นๆ บนตารางได้
schemaCheckEnabled	"จริง"	ควบคุมว่าจะบังคับใช้การตรวจสอบความถูกต้องของ Schema ที่เข้มงวดระหว่าง Spark DataFrame และตาราง SQL หรือไม่ ค่าที่เป็นไปได้: true (ค่าเริ่มต้น บังคับใช้การจับคู่สคีมาที่เข้มงวด) false (ให้ความยืดหยุ่นมากขึ้นและอาจข้ามการตรวจสอบสคีมาบางอย่าง) การตั้งค่าเป็น false สามารถช่วยในเรื่อง Schema ไม่ตรงกัน แต่อาจนําไปสู่ผลลัพธ์ที่ไม่คาดคิดหากโครงสร้างแตกต่างกันอย่างมีนัยสําคัญ

ตัวเลือก Bulk API อื่นๆ สามารถตั้งค่าเป็นตัวเลือกบน และDataFrameส่งผ่านไปยัง API การคัดลอกจํานวนมากเมื่อเขียน

ตัวอย่างการเขียนและอ่าน

โค้ดต่อไปนี้ใช้การตรวจสอบสิทธิ์ Microsoft Entra ID อัตโนมัติเพื่อแสดงการดําเนินการเหล่านี้:

• เขียน DataFrame: df.write.option("...", "...").mssql("<schema>.<table>").
• อ่านตาราง: spark.read.option("...", "...").mssql("<schema>.<table>").
• รันคําสั่งที่กําหนดเอง: spark.read.option("...", "...").option("query", "<your-custom-query>").mssql().

เคล็ดลับ

ข้อมูลถูกสร้างขึ้นแบบอินไลน์เพื่อวัตถุประสงค์ในการสาธิต ในสถานการณ์การใช้งานจริง โดยทั่วไป คุณจะอ่านข้อมูลจากแหล่งข้อมูลที่มีอยู่หรือสร้าง DataFrame.

PySpark

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)
df.write.mode("overwrite").option("url", url).mssql("dbo.publicExample")
spark.read.option("url", url).mssql("dbo.publicExample").show()
spark.read.option("url", url).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

url = "jdbc:sqlserver://<server>:<port>;database=<database2>;" # different database
df.write.mode("overwrite").option("url", url).mssql("dbo.tableInDatabase2") # default url is updated
spark.read.mssql("dbo.tableInDatabase2").show() # no url option specified and will use database2

สเกลาสปาร์ค

import com.microsoft.sqlserver.jdbc.spark.SparkSqlImplicits._
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
val url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
val row_data = Seq(
  Row("Alice", 1),
  Row("Bob", 2),
  Row("Charlie", 3)
)
val schema = List(
  StructField("Name", StringType, nullable = true),
  StructField("Age", IntegerType, nullable = true)
)
val df = spark.createDataFrame(spark.sparkContext.parallelize(row_data), StructType(schema))
df.write.mode("overwrite").option("url", url).mssql("dbo.publicExample")
spark.read.option("url", url).mssql("dbo.publicExample").show
spark.read.option("url", url).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show // Read with a custom query

val url = "jdbc:sqlserver://<server>:<port>;database=<database2>;" // different database
df.write.mode("overwrite").option("url", url).mssql("dbo.tableInDatabase2") // default url is updated
spark.read.mssql("dbo.tableInDatabase2").show // no url option specified and will use database2

คุณยังสามารถเลือกคอลัมน์ ใช้ตัวกรอง และใช้ตัวเลือกอื่นๆ เมื่อคุณอ่านข้อมูลจากกลไกจัดการฐานข้อมูล SQL

ตัวอย่างการรับรองความถูกต้อง

ตัวอย่างต่อไปนี้แสดงวิธีการใช้วิธีการรับรองความถูกต้องอื่นที่ไม่ใช่ Microsoft Entra ID เช่น บริการหลัก (โทเค็นการเข้าถึง) และการรับรองความถูกต้อง SQL

Note

ดังที่ได้กล่าวไว้ก่อนหน้านี้ การรับรองความถูกต้องของ Microsoft Entra ID จะได้รับการจัดการโดยอัตโนมัติเมื่อคุณลงชื่อเข้าใช้พื้นที่ทํางาน Fabric ดังนั้นคุณต้องใช้วิธีการเหล่านี้หากสถานการณ์ของคุณต้องการเท่านั้น

บริการหลักหรือโทเค็นการเข้าถึง

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)

from azure.identity import ClientSecretCredential
credential = ClientSecretCredential(tenant_id="", client_id="", client_secret="") # service principal app
scope = "https://database.windows.net/.default"
token = credential.get_token(scope).token

df.write.mode("overwrite").option("url", url).option("accesstoken", token).mssql("dbo.publicExample")
spark.read.option("accesstoken", token).mssql("dbo.publicExample").show()
spark.read.option("accesstoken", token).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

ผู้ใช้/รหัสผ่าน

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)
df.write.mode("overwrite").option("url", url).option("user", "").option("password", "").mssql("dbo.publicExample")
spark.read.option("user", "").option("password", "").mssql("dbo.publicExample").show()
spark.read.option("user", "").option("password", "").option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

โหมดบันทึก DataFrame ที่รองรับ

เมื่อคุณเขียนข้อมูลจากฐานข้อมูล Spark ไปยัง SQL คุณสามารถเลือกโหมดบันทึกได้หลายโหมด โหมดบันทึกจะควบคุมวิธีการเขียนข้อมูลเมื่อมีตารางปลายทางอยู่แล้ว และอาจส่งผลต่อ Schema ข้อมูล และการจัดทําดัชนี การทําความเข้าใจโหมดเหล่านี้จะช่วยให้คุณหลีกเลี่ยงการสูญหายหรือการเปลี่ยนแปลงข้อมูลที่ไม่คาดคิด

ตัวเชื่อมต่อนี้รองรับตัวเลือกที่กําหนดไว้ที่นี่: ฟังก์ชัน Spark Save

• ErrorIfExists (โหมดบันทึกดีฟอนต์): ถ้ามีตารางปลายทางอยู่ การเขียนจะถูกยกเลิกและส่งกลับข้อยกเว้น มิฉะนั้น ตารางใหม่จะถูกสร้างขึ้นด้วยข้อมูล

• ละเว้น: หากมีตารางปลายทางอยู่ การเขียนจะละเว้นคําขอและไม่ส่งคืนข้อผิดพลาด มิฉะนั้น ตารางใหม่จะถูกสร้างขึ้นด้วยข้อมูล

• เขียนทับ: ถ้ามีตารางปลายทางอยู่ ตารางจะถูกทิ้ง สร้างใหม่ และผนวกข้อมูลใหม่

  Note

  เมื่อคุณใช้ overwrite, คุณจะสูญเสียสคีมาตารางต้นฉบับ (โดยเฉพาะชนิดข้อมูลเฉพาะของ MSSQL) และดัชนีตาราง สคีมาจะถูกแทนที่ด้วยสคีมาที่อนุมานได้จาก Spark DataFrame ของคุณ เพื่อหลีกเลี่ยงการสูญเสียสคีมาและดัชนี ให้เพิ่ม.option("truncate", true)

• ผนวก: ถ้ามีตารางปลายทางอยู่ จะมีการผนวกข้อมูลใหม่เข้าไป มิฉะนั้น ตารางใหม่จะถูกสร้างขึ้นด้วยข้อมูล

แก้ไขปัญหา

เมื่อกระบวนการเสร็จสิ้น ผลลัพธ์ของการดําเนินการอ่าน Spark ของคุณจะปรากฏในพื้นที่เอาต์พุตของเซลล์ ข้อผิดพลาดจาก com.microsoft.sqlserver.jdbc.SQLServerException มาจาก SQL Server โดยตรง คุณสามารถค้นหาข้อมูลข้อผิดพลาดโดยละเอียดได้ในบันทึกแอปพลิเคชัน Spark

การเขียนแบบกลุ่มต้องการให้ข้อมูลขาเข้าตรงกับชนิดข้อมูล SQL เป้าหมาย หากข้อมูลไม่ตรงกัน คุณอาจได้รับข้อผิดพลาดนี้:

Caused by: com.microsoft.sqlserver.jdbc.SQLServerException: The service has encountered an error processing your request. Please try again. Error code 4815.

ตัวอย่างเช่น ข้อผิดพลาดนี้อาจเกิดขึ้นเมื่อตาราง SQL เป้าหมายใช้ datetime, แต่ค่า Spark TimestampType ที่เข้ามา เช่น 2025-01-01 10:30:00.123456, มีความแม่นยํามากกว่า datetime ที่รองรับ

เพื่อแก้ไขข้อผิดพลาด ให้ใช้วิธีใดวิธีหนึ่งต่อไปนี้:

• แปลง ตัดทอน หรือแปลงข้อมูลที่เข้ามาให้ตรงกับชนิดข้อมูล SQL เป้าหมาย ตัวอย่างเช่น ตัดค่าให้เหลือสามหลักที่มีความแม่นยําแบบเศษส่วนวินาที: 2025-01-01 10:30:00.123000
• อนุญาตให้ตัวเชื่อมต่อสร้างสคีมาของตารางใหม่โดยการตั้งค่า .option("truncate", false). ตัวเชื่อมต่อจะแมปประเภท timestamp ของ Spark เป็น SQL datetime2

เนื้อหาที่เกี่ยวข้อง

• ฐานข้อมูล Azure SQL
• ฐานข้อมูล SQL ใน Fabric
• ฐานข้อมูล Azure SQL - การรับรองความถูกต้องและการอนุญาต