Synapse Spark能否连接Log Analytics workspace及数据导出至存储账户咨询

1. Synapse Spark是否可以连接Log Analytics Workspace？ #

可以连接，主要有三种实现路径：

• 通过Log Analytics REST API结合Spark的HTTP客户端发起查询请求，获取数据后进行处理；
• 将Log Analytics Workspace（LAW）链接到Azure Data Explorer（ADX）集群，使用ADX Spark连接器访问LAW数据；
• 利用Azure Monitor数据导出功能先将LAW数据同步到存储账户，再通过Synapse Spark读取存储中的数据。

2. 将Log Analytics Workspace数据导出至存储账户（Parquet/Delta格式）的实现方法 #

方法一：Azure Monitor数据导出功能（直接导出Parquet） #

• 操作步骤：
  • 登录Azure门户，进入目标Log Analytics Workspace，找到「数据导出」选项；
  • 创建导出规则，指定要导出的日志表、目标存储账户（支持ADLS Gen2或Blob存储），并选择导出格式为Parquet；
  • 规则生效后，LAW会自动将新增数据同步到存储账户的指定路径；
  • 若需转换为Delta格式，可在Synapse Notebook中读取存储中的Parquet文件，再写入Delta表：

    # 读取存储中的Parquet数据
    df = spark.read.parquet("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<LAW导出路径>")
    # 写入Delta格式到存储账户
    df.write.format("delta").mode("append").save("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<Delta存储路径>")
    

方法二：Synapse Spark调用Log Analytics API读取后写入存储 #

• 核心逻辑：通过Spark调用LAW的查询API（使用Kusto查询语言）获取数据，直接写入存储的Parquet/Delta格式。示例代码（PySpark）：

  import requests
  from pyspark.sql import SparkSession
  
  # 获取Azure AD访问令牌（用于LAW API认证）
  def get_access_token():
      token_url = "https://login.microsoftonline.com/<租户ID>/oauth2/token"
      payload = {
          "grant_type": "client_credentials",
          "client_id": "<服务主体ID>",
          "client_secret": "<服务主体密钥>",
          "resource": "https://api.loganalytics.io"
      }
      response = requests.post(token_url, data=payload)
      return response.json()["access_token"]
  
  # 调用LAW查询API获取数据
  token = get_access_token()
  query_url = "https://api.loganalytics.io/v1/workspaces/<LAW工作区ID>/query"
  headers = {"Authorization": f"Bearer {token}"}
  query_payload = {"query": "AzureActivity | take 100"}  # 替换为你的Kusto查询
  response = requests.post(query_url, headers=headers, json=query_payload)
  query_results = response.json()["tables"][0]["rows"]
  columns = [col["name"] for col in response.json()["tables"][0]["columns"]]
  
  # 转换为Spark DataFrame并写入存储
  df = spark.createDataFrame(query_results, schema=columns)
  # 写入Parquet
  df.write.parquet("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<Parquet存储路径>", mode="overwrite")
  # 写入Delta
  df.write.format("delta").mode("overwrite").save("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<Delta存储路径>")
  

方法三：通过ADX集群链接LAW后导出 #

• 操作步骤：
  1. 将LAW链接到ADX集群：在Azure门户的LAW中进入「链接的服务」，添加Azure Data Explorer集群，完成关联；
  2. 在Synapse Notebook中使用ADX Spark连接器读取LAW数据（此时LAW作为ADX的一个数据库）；
  3. 将读取的数据写入存储账户的Parquet/Delta格式：

  # 配置ADX Spark连接器参数
  spark.conf.set("spark.kusto.clusterUrl", "https://<ADX集群名>.<区域>.kusto.windows.net")
  spark.conf.set("spark.kusto.database", "<LAW工作区名称>")
  spark.conf.set("spark.kusto.authenticationMethod", "AADServicePrincipal")
  spark.conf.set("spark.kusto.aadTenantId", "<租户ID>")
  spark.conf.set("spark.kusto.clientId", "<服务主体ID>")
  spark.conf.set("spark.kusto.clientSecret", "<服务主体密钥>")
  
  # 读取LAW中的数据
  df = spark.read.format("com.microsoft.kusto.spark.synapse.datasource") \
      .option("kustoDatabase", "<LAW工作区名称>") \
      .option("kustoQuery", "AzureActivity | take 100") \
      .load()
  
  # 写入Parquet/Delta到存储
  df.write.parquet("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<路径>", mode="overwrite")
  df.write.format("delta").mode("overwrite").save("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<路径>")
  

3. 使用ADX Spark连接器时URL无效报错的解决方法 #

ADX Spark连接器仅支持Azure Data Explorer集群的URL，无法直接使用Log Analytics Workspace的URL，解决步骤如下：

• 确认你的LAW已链接到ADX集群：若未链接，需在Azure门户的LAW「链接的服务」中添加ADX集群，完成关联；
• 获取ADX集群的正确URL：格式为 https://<ADX集群名称>.<区域>.kusto.windows.net，可在ADX集群的Azure门户页面中找到；
• 在Synapse Notebook中修改连接器配置，使用上述ADX集群URL，同时指定LAW对应的数据库名称（即LAW的工作区名称），示例配置参考方法三中的代码。

内容的提问来源于stack exchange，提问作者Nosferatus2K