标签: databricks

Sql Delta 表和 Databricks Delta 表之间有区别吗？看起来对于 SQL，我们在概念上使用这个名称。存储Base表差异的表是Delta。数据块也一样吗？

我有兴趣创建一个托管在 Databricks 笔记本中的交互式图表。我一直在尝试 Plotly，但我不知道如何让任何自定义交互功能发挥作用。具体来说，我需要一个点击事件处理程序。我使用的以下代码基于Plotly 文档中的示例。唯一的区别是最后一行，我用它displayHTML()来渲染图形。

import plotly.graph_objects as go

import numpy as np
np.random.seed(1)

x = np.random.rand(100)
y = np.random.rand(100)

f = go.FigureWidget([go.Scatter(x=x, y=y, mode='markers')])

scatter = f.data[0]
colors = ['#a3a7e4'] * 100
scatter.marker.color = colors
scatter.marker.size = [10] * 100
f.layout.hovermode = 'closest'


# create our callback function
def update_point(trace, points, selector):
    c = list(scatter.marker.color)
    s = list(scatter.marker.size)
    for i in points.point_inds:
        c[i] = '#bae2be'
        s[i] = 20
        with f.batch_update():
            scatter.marker.color = c
            scatter.marker.size = …

我无法通过门户将 Azure AD 服务主体名称添加到 Azure Databricks 中，但我能够在 Databricks API 端点的帮助下添加我的服务主体。如何为我的服务主体名称创建 PAT 令牌。

希望快点；

将一些本地 SQL 视图迁移到 Databricks 并努力寻找某些函数的转换。主要函数是 string_agg 函数。

string_agg(字段名称, ', ')

有人知道如何将其转换为 Databricks SQL 吗？

提前致谢。

我多次在 Databricks 中更新增量表时遇到问题，第一次覆盖架构失败，但第二次成功。我的问题的解决方案是简单地再次运行它，但此时我无法重现。如果再次发生这种情况，我会回来并发布确切的错误消息，但这本质上是架构不匹配错误。还有其他人遇到过类似的问题吗？

overwriteSchema = True
DF.write \
.format("delta") \
.mode("overwrite") \
.option("overwriteSchema", overwriteSchema) \
.partitionBy(datefield) \
.saveAsTable(deltatable)

Databricks文档中的所有示例均采用 Scala 语言。无法从 PySpark 找到如何使用此触发器类型。是否有等效的 API 或解决方法？

DELTA不具有CREATE TABLE LIKE。它确实有CTAS。

我只想复制表的定义LOCATION，但还要指定.

例如，这不起作用：

CREATE TABLE IF NOT EXISTS NEW_CUSTOMER_FEED 
AS SELECT * from NEW_CUSTOMER_FEED WHERE 1 = 0 
LOCATION '/atRest/data'

我缺少什么？

我可以迭代 databricks 笔记本中的小部件吗？像这样的伪代码？

# NB - not valid
inputs = {widget.name: widget.value for widget in dbutils.widgets}

由于多次连接，我有一个数据框。我想调查重复的情况。但每次我调查时，数据框看起来都不一样。特别是，以下命令会导致不同IDs但结果数量保持不变。

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
import pyspark.sql.functions as f
from pyspark.sql.functions import lit

# Create a Spark session
spark = SparkSession.builder.appName("CreateDataFrame").getOrCreate()

# User input for number of rows
n_a = 10
n_a_c = 5
n_a_c_d = 3
n_a_c_e = 4

# Define the schema for the DataFrame
schema_a = StructType([StructField("id1", StringType(), True)])
schema_a_b = StructType(
    [
        StructField("id1", StringType(), True),
        StructField("id2", StringType(), True),
        StructField("extra", StringType(), True),
    ]
)
schema_a_c = StructType(
    [ …

到底如何创建一个高效且可重用的 Databricks 工作流程来将原始 SQL 数据库转储到 Delta Lake 中。这里的一些混淆是为了实现以下目的的最佳方法：

• 处理模式中的偏差（数据库表中的列）=> 对存储的表进行简单的覆盖可以吗？
• 捕获数据变化（CDC）并高效合并现有数据；身份证上说。这对于关系数据库是否仍然相关？
• Delta Live Table (DLT) 格式适合这个吗？

人们可能会想象以下过程：

1. 迭代公共表 information_schema：

table_names = spark.read.jdbc(url=jdbcUrl, table="information_schema.tables",
                               properties=connectionProperties) \
                               .filter("table_schema = 'public'") \
                               .select("table_name") \
                               .rdd.flatMap(lambda x: x) \
                               .collect()

for table in table_names:
    ...

2. 然后对于每个表：

• (A) 创建一个新的 Delta Lake 表，如果它不存在（或者可能在架构方面已经过时），否则；
• (B) 将新数据/更新数据合并到 Delta Lake 中。

像 Airbyte 和其他公司这样的第三方供应商提供了这项服务——并不是因为它确实应该如此难以实施。但更有可能的是，由于 Databricks DLT/Delta Lake 方面此通用流程的文档或参考实现乏善可陈。

令人满意的答案将是（I）对 OP 中包含的（错误？）假设的一些背景/验证，（II）此工作流程缺少的代码，以及（III）对提出的 3 点的答案/澄清。