Azure Synapse Analytics에서 데이터 쿼리

Azure Synapse에서 임시 스테이징에 Azure Data Lake Storage Gen2 스토리지 계정을 사용하여 Azure Databricks 클러스터와 Azure Synapse 인스턴스 간에 대량의 데이터를 효율적으로 전송하는 Azure Synapse 커넥터(Azure Synapse에서 COPY 문 사용)를 사용하여 Azure Databricks의 Azure Synapse에 액세스할 수 있습니다.

Azure Synapse Analytics는 MPP(대규모 병렬 처리)를 활용하는 클라우드 기반 엔터프라이즈 데이터 웨어하우스이며 페타바이트 데이터에서 복잡한 쿼리를 신속하게 실행합니다.

Synapse에 대한 예제 구문

Scala, Python, SQL 및 R에서 Synapse를 쿼리할 수 있습니다. 다음 코드 예제에서는 스토리지 계정 키를 사용하고 Azure Databricks에서 Synapse로 스토리지 자격 증명을 전달합니다.

Databricks는 사용 가능한 가장 안전한 인증 흐름을 사용하는 것이 좋습니다. 이 예제에 설명된 인증 흐름은 다른 흐름에 없는 위험을 수반합니다. 관리 ID와 같은 다른 보안 흐름이 실행 가능하지 않은 경우에만 이 흐름을 사용해야 합니다.

Scala

// Set up the storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.dfs.core.windows.net",
  "<your-storage-account-access-key>")

// Get some data from an Azure Synapse table. The following example applies to Databricks Runtime 11.3 LTS and above.
val df: DataFrame = spark.read
  .format("sqldw")
  .option("host", "hostname")
  .option("port", "port") /* Optional - will use default port 1433 if not specified. */
  .option("user", "username")
  .option("password", "password")
  .option("database", "database-name")
  .option("dbtable", "schema-name.table-name") /* If schemaName not provided, default to "dbo". */
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .load()

// Get some data from an Azure Synapse table. The following example applies to Databricks Runtime 10.4 LTS and below.
val df: DataFrame = spark.read
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("dbTable", "<your-table-name>")
  .load()

// Load data from an Azure Synapse query.
val df: DataFrame = spark.read
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("query", "select x, count(*) as cnt from table group by x")
  .load()

// Apply some transformations to the data, then use the
// Data Source API to write the data back to another table in Azure Synapse.

df.write
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("dbTable", "<your-table-name>")
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")
  .save()

Python

# Set up the storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.dfs.core.windows.net",
  "<your-storage-account-access-key>")

# Get some data from an Azure Synapse table. The following example applies to Databricks Runtime 11.3 LTS and above.
df = spark.read
  .format("sqldw")
  .option("host", "hostname")
  .option("port", "port") # Optional - will use default port 1433 if not specified.
  .option("user", "username")
  .option("password", "password")
  .option("database", "database-name")
  .option("dbtable", "schema-name.table-name") # If schemaName not provided, default to "dbo".
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .load()

# Get some data from an Azure Synapse table. The following example applies to Databricks Runtime 10.4 LTS and below.
df = spark.read \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "<your-table-name>") \
  .load()

# Load data from an Azure Synapse query.
df = spark.read \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("query", "select x, count(*) as cnt from table group by x") \
  .load()

# Apply some transformations to the data, then use the
# Data Source API to write the data back to another table in Azure Synapse.

df.write \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "<your-table-name>") \
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>") \
  .save()

SQL

-- Set up the storage account access key in the notebook session conf.
SET fs.azure.account.key.<your-storage-account-name>.dfs.core.windows.net=<your-storage-account-access-key>;

-- Read data using SQL. The following example applies to Databricks Runtime 11.3 LTS and above.
CREATE TABLE example_table_in_spark_read
USING sqldw
OPTIONS (
  host '<hostname>',
  port '<port>' /* Optional - will use default port 1433 if not specified. */
  user '<username>',
  password '<password>',
  database '<database-name>'
  dbtable '<schema-name>.<table-name>', /* If schemaName not provided, default to "dbo". */
  forwardSparkAzureStorageCredentials 'true',
  tempDir 'abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>'
);

-- Read data using SQL. The following example applies to Databricks Runtime 10.4 LTS and below.
CREATE TABLE example_table_in_spark_read
USING com.databricks.spark.sqldw
OPTIONS (
  url 'jdbc:sqlserver://<the-rest-of-the-connection-string>',
  forwardSparkAzureStorageCredentials 'true',
  dbtable '<your-table-name>',
  tempDir 'abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>'
);

-- Write data using SQL.
-- Create a new table, throwing an error if a table with the same name already exists:

CREATE TABLE example_table_in_spark_write
USING com.databricks.spark.sqldw
OPTIONS (
  url 'jdbc:sqlserver://<the-rest-of-the-connection-string>',
  forwardSparkAzureStorageCredentials 'true',
  dbTable '<your-table-name>',
  tempDir 'abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>'
)
AS SELECT * FROM table_to_save_in_spark;

R

# Load SparkR
library(SparkR)

# Set up the storage account access key in the notebook session conf.
conf <- sparkR.callJMethod(sparkR.session(), "conf")
sparkR.callJMethod(conf, "set", "fs.azure.account.key.<your-storage-account-name>.dfs.core.windows.net", "<your-storage-account-access-key>")

# Get some data from an Azure Synapse table.
df <- read.df(
   source = "com.databricks.spark.sqldw",
   url = "jdbc:sqlserver://<the-rest-of-the-connection-string>",
   forward_spark_azure_storage_credentials = "true",
   dbTable = "<your-table-name>",
   tempDir = "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")

# Load data from an Azure Synapse query.
df <- read.df(
   source = "com.databricks.spark.sqldw",
   url = "jdbc:sqlserver://<the-rest-of-the-connection-string>",
   forward_spark_azure_storage_credentials = "true",
   query = "select x, count(*) as cnt from table group by x",
   tempDir = "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")

# Apply some transformations to the data, then use the
# Data Source API to write the data back to another table in Azure Synapse.

write.df(
  df,
  source = "com.databricks.spark.sqldw",
  url = "jdbc:sqlserver://<the-rest-of-the-connection-string>",
  forward_spark_azure_storage_credentials = "true",
  dbTable = "<your-table-name>",
  tempDir = "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.windows.net/<your-directory-name>")

Azure Databricks와 Synapse 간의 인증은 어떻게 작동하나요?

Azure Synapse 커넥터는 다음 세 가지 유형의 네트워크 연결을 사용합니다.

• Azure Synapse에 대한 Spark 드라이버
• Spark 클러스터에서 Azure Storage 계정으로
• Azure 스토리지 계정에 대한 Azure Synapse

Azure Storage에 대한 액세스 구성

Azure Databricks와 Synapse는 모두 임시 데이터 스토리지에 사용할 Azure Storage 계정에 대한 권한 있는 액세스가 필요합니다.

Azure Synapse는 스토리지 계정 액세스에 SAS 사용을 지원하지 않습니다. 다음 중 하나를 수행하여 두 서비스에 대한 액세스를 구성할 수 있습니다.

• 스토리지 계정의 계정 키와 비밀을 사용하고 forwardSparkAzureStorageCredentials를 true로 설정합니다. Azure Storage에 액세스 가능한 Azure 자격 증명을 구성하도록 Spark 속성 설정을 참조하세요.
• OAuth 2.0 인증에 Azure Data Lake Storage Gen2를 사용하고 enableServicePrincipalAuth를 true로 설정합니다. 서비스 주체를 사용하여 OAuth 2.0으로 Azure Databricks에서 Synapse로의 연결 구성을 참조하세요.
• 관리되는 서비스 ID를 갖도록 Azure Synapse 인스턴스를 구성하고 useAzureMSI를 true로 설정합니다.

필요한 Azure Synapse 권한

이는 백그라운드에서 COPY를 사용하므로 Azure Synapse 커넥터를 사용하려면 JDBC 연결 사용자에게 연결된 Azure Synapse 인스턴스에서 다음 명령을 실행할 수 있는 권한이 있어야 합니다.

• COPY INTO

대상 테이블이 Azure Synapse에 없는 경우 위의 명령 외에도 다음 명령을 실행할 수 있는 권한이 필요합니다.

• CREATE TABLE

다음 표에는 COPY를 사용한 쓰기에 필요한 권한이 요약되어 있습니다.

네트워킹 구성

Azure Synapse에서 방화벽을 구성하는 경우 Azure Databricks가 Azure Synapse에 연결할 수 있도록 네트워크 설정을 구성해야 합니다. 먼저 Azure Virtual Network에 Azure Databricks 배포(VNet 삽입)에 따라 Azure Databricks 작업 영역이 자체 가상 네트워크에 배포되었는지 확인합니다. 그런 다음, 서브넷에서 Synapse 계정으로의 연결을 허용하도록 Azure Synapse에서 IP 방화벽 규칙을 구성할 수 있습니다. Azure Synapse Analytics IP 방화벽 규칙을 참조하세요.

서비스 주체를 사용하여 OAuth 2.0으로 Azure Databricks에서 Synapse로의 연결 구성

기본 스토리지 계정에 대한 액세스 권한이 있는 서비스 주체를 사용하여 Azure Synapse Analytics에서 인증을 받을 수 있습니다. 서비스 주체 자격 증명을 사용하여 Azure 스토리지 계정에 액세스하는 방법에 대한 자세한 내용은 Azure Data Lake Storage Gen2 및 Blob Storage에 액세스를 참조하세요. 커넥터가 서비스 주체를 사용하여 인증할 수 있도록 하려면 연결 구성 enableServicePrincipalAuth에서 true 옵션을 로 설정해야 합니다.

필요에 따라 Azure Synapse Analytics 연결에 다른 서비스 주체를 사용할 수 있습니다. 다음 예제에서는 스토리지 계정에 대한 서비스 주체 자격 증명 및 Synapse에 대한 선택적 서비스 주체 자격 증명을 구성합니다.

ini

; Defining the Service Principal credentials for the Azure storage account
fs.azure.account.auth.type OAuth
fs.azure.account.oauth.provider.type org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider
fs.azure.account.oauth2.client.id <application-id>
fs.azure.account.oauth2.client.secret <service-credential>
fs.azure.account.oauth2.client.endpoint https://login.microsoftonline.com/<directory-id>/oauth2/token

; Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
spark.databricks.sqldw.jdbc.service.principal.client.id <application-id>
spark.databricks.sqldw.jdbc.service.principal.client.secret <service-credential>

Scala

// Defining the Service Principal credentials for the Azure storage account
spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type",  "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id", "<application-id>")
spark.conf.set("fs.azure.account.oauth2.client.secret", "<service-credential>")
spark.conf.set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<directory-id>/oauth2/token")

// Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
spark.conf.set("spark.databricks.sqldw.jdbc.service.principal.client.id", "<application-id>")
spark.conf.set("spark.databricks.sqldw.jdbc.service.principal.client.secret", "<service-credential>")

Python

# Defining the service principal credentials for the Azure storage account
spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type",  "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id", "<application-id>")
spark.conf.set("fs.azure.account.oauth2.client.secret", "<service-credential>")
spark.conf.set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<directory-id>/oauth2/token")

# Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
spark.conf.set("spark.databricks.sqldw.jdbc.service.principal.client.id", "<application-id>")
spark.conf.set("spark.databricks.sqldw.jdbc.service.principal.client.secret", "<service-credential>")

R

# Load SparkR
library(SparkR)
conf <- sparkR.callJMethod(sparkR.session(), "conf")

# Defining the service principal credentials for the Azure storage account
sparkR.callJMethod(conf, "set", "fs.azure.account.auth.type", "OAuth")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth.provider.type",  "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth2.client.id", "<application-id>")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth2.client.secret", "<service-credential>")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<directory-id>/oauth2/token")

# Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
sparkR.callJMethod(conf, "set", "spark.databricks.sqldw.jdbc.service.principal.client.id", "<application-id>")
sparkR.callJMethod(conf, "set", "spark.databricks.sqldw.jdbc.service.principal.client.secret", "<service-credential>")

일괄 쓰기에 지원되는 저장 모드

Azure Synapse 커넥터는 기본 모드가 ErrorIfExists인 Ignore, Append, Overwrite 및 ErrorIfExists 저장 모드를 지원합니다. Apache Spark에서 지원되는 저장 모드에 대한 자세한 내용은 Spark SQL 설명서의 저장 모드를 참조하세요.

Azure Databricks Synapse 커넥터 옵션 참조

Spark SQL에 제공된 OPTIONS는 다음 설정을 지원합니다.

Azure Synapse Analytics로 쿼리 푸시다운

Azure Synapse 커넥터는 다음 연산자를 Azure Synapse로 푸시하는 최적화 규칙 집합을 구현합니다.

• Filter
• Project
• Limit

Project 및 Filter 연산자는 다음 식을 지원합니다.

• 대부분의 부울 논리 연산자
• 비교
• 기본 산술 연산
• 숫자 및 문자열 캐스트

Limit 연산자의 경우 지정된 순서가 없는 경우에만 푸시다운이 지원됩니다. 예시:

SELECT TOP(10) * FROM table이지만, SELECT TOP(10) * FROM table ORDER BY col이 아닙니다.

Azure Synapse 커넥터를 사용하여 빌드된 쿼리 푸시다운은 기본적으로 사용하도록 설정됩니다. spark.databricks.sqldw.pushdown을 false로 설정하여 사용하지 않도록 설정할 수 있습니다.

임시 데이터 관리

Azure Synapse 커넥터는 Azure Storage 컨테이너에서 만든 임시 파일을 삭제하지 않습니다. 사용자가 제공한 tempDir 위치에서 임시 파일을 주기적으로 삭제하는 것이 좋습니다.

데이터 정리를 용이하게 하기 위해 Azure Synapse 커넥터는 데이터 파일을 tempDir 바로 아래에 저장하지 않고 대신 <tempDir>/<yyyy-MM-dd>/<HH-mm-ss-SSS>/<randomUUID>/ 형식의 하위 디렉터리를 만듭니다. 제공된 임계값보다 오래 실행되는 Spark 작업은 있을 수 없다고 가정하고, 제공된 임계값보다 오래된(예: 2일) 하위 디렉터리를 재귀적으로 삭제하도록 주기적 작업(Azure Databricks 작업 기능 사용 여부와 관계 없이)을 설정할 수 있습니다.

더 간단한 대안은 전체 컨테이너를 주기적으로 삭제하고 이름이 같은 새 컨테이너를 만드는 것입니다. 이렇게 하려면 Azure Synapse 커넥터에서 생성한 임시 데이터에 전용 컨테이너를 사용해야 하며 커넥터와 관련된 쿼리가 실행되고 있지 않음을 보장할 수 있는 기간을 찾을 수 있어야 합니다.

임시 개체 관리

Azure Synapse 커넥터는 Azure Databricks 클러스터와 Azure Synapse 인스턴스 간의 데이터 전송을 자동화합니다. Azure Synapse 테이블 또는 쿼리에서 데이터를 읽거나 Azure Synapse 테이블에 데이터를 쓰기 위해 Azure Synapse 커넥터는 내부적으로 DATABASE SCOPED CREDENTIAL, EXTERNAL DATA SOURCE, EXTERNAL FILE FORMAT 및 EXTERNAL TABLE을 포함하여 임시 개체를 만듭니다. 이러한 개체는 해당 Spark 작업 기간 동안에만 활성 상태가 되고 자동으로 삭제됩니다.

클러스터가 Azure Synapse 커넥터를 사용하여 쿼리를 실행 중일 때 Spark 드라이버 프로세스가 충돌하거나 강제로 다시 시작되는 경우 또는 클러스터가 강제로 종료되거나 다시 시작되면 임시 개체가 삭제되지 않을 수 있습니다. 이러한 개체의 식별 및 수동 삭제를 용이하게 하기 위해 Azure Synapse 커넥터는 Azure Synapse 인스턴스에서 만든 모든 중간 임시 개체의 이름 앞에 tmp_databricks_<yyyy_MM_dd_HH_mm_ss_SSS>_<randomUUID>_<internalObject> 형식의 태그를 추가합니다.

다음과 같은 쿼리를 사용하여 유출된 개체를 주기적으로 찾는 것이 좋습니다.

• SELECT * FROM sys.database_scoped_credentials WHERE name LIKE 'tmp_databricks_%'
• SELECT * FROM sys.external_data_sources WHERE name LIKE 'tmp_databricks_%'
• SELECT * FROM sys.external_file_formats WHERE name LIKE 'tmp_databricks_%'
• SELECT * FROM sys.external_tables WHERE name LIKE 'tmp_databricks_%'