Azure Database ricks：为谁在什么时刻运行了什么查询创建审计跟踪

Question

我们有一个审计需求，可以洞察Azure Databricks中谁在什么时候执行了什么查询。Azure Databricks / Spark UI / Jobs选项卡已经列出了执行的Spark作业，包括完成的查询和提交的时间。但是它不包括谁执行了查询。

1. 是否有我们可以与Azure Databricks一起使用的API来查询UI中显示的这些Spark作业详细信息？（Databricks REST API似乎没有提供这一点，但可能我忽略了一些东西）
2. 我们是否可以确定谁创建了Spark作业（使用API）

谢谢，下吕

Answer 1

根据Douglas的回答，我想出了这个函数来在Databricks笔记本中使用，并获得有关缓存RDD的一些信息：

我希望它有帮助。我正在寻找关于这个主题的2h的信息。

def get_databricks_rdd_info():

    import requests, json

    # Get Spark Context
    sc = spark.sparkContext
    # Get App Id (Notebook is attached to it)
    app_id = sc._jsc.sc().applicationId()
    # Where is my driver
    driver_ip = spark.conf.get('spark.driver.host')
    port = spark.conf.get("spark.ui.port")
    # Compose the query to the Spark UI API
    url = f"http://{driver_ip}:{port}/api/v1/applications/{app_id}/storage/rdd"

    # Make request
    r = requests.get(url, timeout=3.0)

    if r.status_code == 200:
        # Compose results
        df = spark.createDataFrame([json.dumps(r) for r in r.json()], T.StringType())
        json_schema = spark.read.json(df.rdd.map(lambda row: row.value)).schema
        df = df.withColumn('value', F.from_json(F.col('value'), json_schema))
        df = df.selectExpr('value.*')
        
        # Generate summary
        df_summary = (df
                      .withColumn('Name', F.element_at(F.split(F.col('name'), ' '), -1))
                      .withColumn('Cached', F.round(F.lit(100) * F.col('numCachedPartitions')/F.col('numPartitions'), 2))
                      .withColumn('Memory GB', F.round(F.col('memoryUsed')*1e-9, 2))
                      .withColumn('Disk GB', F.round(F.col('diskUsed')*1e-9, 2))
                      .withColumnRenamed('numPartitions', '# Partitions')

                      .select([
                          'Name',
                          'id',
                          'Cached',
                          'Memory GB',
                          'Disk GB',
                          '# Partitions',
                      ]))
    else:
        print('Some error happened, code:', r.status_code)
        df = None
        df_summary = None
        
        
    return df, df_summary

您可以在数据砖笔记本中将其用作：

df_rdd_info, df_summary = get_databricks_rdd_info()
display(df_summary)

Answer 2

一个。驱动程序节点（内部）访问 Azure Databricks Spark api：

import requests

driverIp = spark.conf.get('spark.driver.host')
port = spark.conf.get("spark.ui.port")
url = F"http://{driverIp}:{port}/api/v1/applications"
r = requests.get(url, timeout=3.0)
r.status_code, r.text

例如，如果您从公共API收到以下错误消息:< code>PERMISSION_DENIED:不允许此端口上的流量

b. 对 Azure 数据砖 Spark API 的外部访问：

import requests
import json
"""
  Program access to Databricks Spark UI. 
  
  Works external to Databricks environment or running within.
  Requires a Personal Access Token. Treat this like a password, do not store in a notebook. Please refer to the Secrets API.
  This Python code requires F string support.

"""

# https://<databricks-host>/driver-proxy-api/o/0/<cluster_id>/<port>/api/v1/applications/<application-id-from-master-spark-ui>/stages/<stage-id>
port = spark.conf.get("spark.ui.port")
clusterId = spark.conf.get("spark.databricks.clusterUsageTags.clusterId")
host = "eastus2.azuredatabricks.net"
workspaceId = "999999999999111"  # follows the 'o=' in the databricks URLs or zero
token = "dapideedeadbeefdeadbeefdeadbeef68ee3"  # Personal Access token

url = F"https://{host}/driver-proxy-api/o/{workspaceId}/{clusterId}/{port}/api/v1/applications/?status=running"
r = requests.get(url, auth=("token", token))

# print Application list response
print(r.status_code, r.text)

applicationId = r.json()[0]['id'] # assumes only one response

url = F"https://{host}/driver-proxy-api/o/{workspaceId}/{clusterId}/{port}/api/v1/applications/{applicationId}/jobs"
r = requests.get(url, auth=("token", token))

print(r.status_code, r.json())

集群日志将是您可以查看的地方，但用户标识不在那里。

投票和跟踪这个想法：https://ideas.databricks.com/ideas/DBE-I-313如何进入想法门户：https://docs.databricks.com/ideas.html