引入Databrick摄取：从不同来源轻松高效地将数据摄取到Delta Lake

我们很高兴在公共预览中推出一项新功能–自动加载程序-和一组合作伙伴集成，该集成允许Databricks用户从各种数据源向Delta Lake增量摄取数据。Auto Loader是Apache Spark的优化云文件源，可在新数据到达时从云存储连续有效地加载数据。合作伙伴集成的数据摄取网络使您可以将数百个数据源中的数据直接摄取到Delta Lake中。

汇集所有数据

组织在各种数据源中都有大量的孤立信息。这些可能会有所不同，从数据库（例如Oracle，MySQL，Postgres等）到产品应用程序（Salesforce，Marketo，HubSpot等）。大量的分析用例需要来自这些不同数据源的数据来生成有意义的报告和预测。例如，完整的渠道分析报告将需要来自各种来源的信息，这些信息从枢纽中的潜在客户信息到Postgres数据库中的产品注册事件。

仅将所有数据集中在数据仓库中是一种反模式，因为Python / R库中的机器学习框架将无法有效访问仓库中的数据。由于您的分析用例涵盖从构建简单的SQL报告到更高级的机器学习预测的范围，因此至关重要的是，您必须使用来自所有数据源的数据以开放格式构建一个中央数据湖，并使其可用于各种用例。

自从我们去年将Delta Lake开放源代码以来，成千上万的组织以一种开放的格式构建中央数据湖，比以前更加可靠和高效。Delta Lake on Databricks提供ACID事务和高效的索引编制，这对于为各种访问模式（从BI工具中的即席SQL查询到计划的脱机培训工作）公开数据至关重要。我们称这种为开放式格式的数据构建集中，可靠和有效的单一事实来源的模式称为“ The Lakehouse”，用于从BI到ML的解耦存储和计算的用例。

图1. Delta Lake的通用数据流。数据被加载到摄取表中，在连续的表中进行精炼，然后用于ML和BI用例。

建造湖舍的一项关键挑战是将来自各种来源的所有数据整合在一起。根据您的数据旅程，数据团队有两种常见方案：

• 从第三方获取数据：通常，您在各种内部数据源（从Hubspot到Postgres数据库）中都有有价值的用户数据。您需要为每个连接器编写专用的连接器，以从源中提取数据并将其存储在Delta Lake中。
• 从云存储中提取数据：您已经具有一种将数据从源中拉到云存储中的机制。当新数据到达云存储时，您需要识别此新数据并将其加载到Delta Lake中进行进一步处理。

来自第三方的数据提取

从内部数据源摄取数据需要为每个数据源编写专用的连接器。这可能是在时间和精力上的巨大投资，需要使用源API来构建连接器并将源模式映射到Delta Lake的模式功能。此外，随着源的API和架构的发展，您还需要维护这些连接器。维护问题与您拥有的每个其他数据源都更加复杂。

为了使您的用户更轻松地访问Delta Lake中的所有数据，我们现在与一组数据提取产品合作。这个数据摄取合作伙伴网络已与Databricks建立了本机集成，以将数据直接摄取并存储在Delta Lake中的云存储中。这可以帮助您的数据科学家和分析人员轻松地开始使用来自各种来源的数据。

Azure Databricks客户已经从与Azure Data Factory的集成中受益，可以将各种来源的数据提取到云存储中。我们很高兴地宣布一组新的合作伙伴- Fivetran，Qlik，Infoworks，StreamSets和Syncsort公司 -帮助用户采集数据来自多种来源。我们还将通过合作伙伴Informatica，Segment和Stitch的更多集成来扩展合作伙伴的数据获取网络。

图2.数据摄取合作伙伴和一些流行的数据源的生态系统，您可以通过这些合作伙伴产品将数据提取到Delta Lake。

从云存储中提取数据

ETL工作负载中的常见工作流是：在将新数据降落到云Blob存储上时对其进行增量处理，使其准备好进行分析。但是，很难以低成本，低延迟和最少的DevOps工作量一次准确地从云Blob存储中连续加载数据。

数据进入Delta表后，借助Delta Lake的ACID事务，可以可靠地读取数据。要从Delta表中流式传输数据，您可以使用Delta源（Azure | AWS），该源利用表的事务日志快速识别添加的新文件。

但是，主要瓶颈在于将存储在云存储中的原始文件加载到Delta表中。基于天真的文件的流媒体源（Azure | AWS）通过重复列出云目录并跟踪看到了哪些文件来标识新文件。由于重复列出文件，将越来越多的文件添加到目录中时，成本和延迟都会迅速增加。为了解决此问题，数据团队通常会采用以下解决方法之一：

• 端到端的高数据延迟：尽管数据每隔几分钟到达一次，但是您可以将数据批处理到一个目录中，然后按计划进行处理。使用基于日或小时的分区目录是一种常用技术。这延长了SLA，使数据可供下游使用者使用。
• 手动DevOps方法：要使SLA保持较低水平，您可以选择利用云通知服务和消息队列服务来通知新文件何时到达消息队列，然后处理新文件。这种方法不仅涉及所需云服务的手动设置过程，而且在有多个需要加载数据的ETL作业时，管理起来也会很快变得复杂。此外，重新处理目录中的现有文件还需要手动列出文件并处理它们（除了云通知设置外），从而增加了设置的复杂性。

Auto Loader是一种优化的文件源，它克服了上述所有限制，并为数据团队提供了一种无缝方式，以最小的DevOps努力以低成本和低延迟加载原始数据。您只需要提供源目录路径并开始流作业即可。新的结构化流媒体源称为“ cloudFiles”，将自动设置文件通知服务，该服务从输入目录订阅文件事件并在新文件到达时对其进行处理，并且还可以选择处理该目录中的现有文件。

图3.以低SLA实现一次准确的数据提取需要手动设置多个云服务。Auto Loader开箱即用地处理所有这些复杂性。

使用自动加载器的主要好处是：

• 无文件状态管理：源将新文件降落到云存储上时，源将对其进行增量处理。您不需要管理有关到达哪些文件的任何状态信息。
• 可扩展：源将通过利用云服务和RocksDB有效地跟踪到达的新文件，而不必列出目录中的所有文件。即使目录中包含数百万个文件，此方法也可扩展。
• 易于使用：源将自动设置增量处理文件所需的通知和消息队列服务。您无需进行任何设置。

使用自动加载程序流式加载

您可以利用Apache Spark熟悉的负载API 开始对流作业进行最少的代码更改：

<code>spark.readStream.format（“ cloudFiles ”）
     .option（“ cloudFiles.format”，“ json”）
     .load（“ / input / path”）
/<code>

使用自动加载程序进行计划的批量加载

如果您每隔几个小时才收到一次数据，则仍可以使用结构化流的Trigger.Once模式在计划的作业中利用自动加载程序。

<code>val df = spark.readStream.format（“ cloudFiles”）
     .option（“ cloudFiles.format”，“ json”）
         .load（“ / input / path”）

df.writeStream.trigger（Trigger.Once）
         .format（“ delta”）
         .start（“ /输出/路径”）
/<code>

您可以使用Databricks Jobs Scheduler（Azure | AWS）将上述代码安排为每小时或每天按计划运行，以增量方式加载新数据。使用上述方法，您无需担心迟到的数据方案。

使用COPY命令计划的批量加载

喜欢使用声明性语法的用户可以使用SQL COPY命令按计划将数据加载到Delta Lake中。COPY命令是幂等的，因此在发生故障时可以安全地重新运行。该命令会自动忽略以前加载的文件，并保证一次语义准确。这使数据团队可以轻松构建健壮的数据管道。

该命令的语法如下所示。有关更多详细信息，请参阅有关COPY命令的文档（Azure | AWS）。

<code>COPY INTO tableIdentifier
来自{ （选择identifierList FROM位置）} 

FILEFORMAT = {CSV | JSON | AVRO | ORC | 派克}
[文件=（''[，''] [，...]）]
[PATTERN ='']
[FORMAT_OPTIONS（'dataSourceReaderOption'='value'，...）]
[COPY_OPTIONS（'force'= {'false'，'true'}）]
/<code>

图4.使用新功能将数据导入Delta Lake。使用Auto Loader进行流式加载可确保一次准确地提取数据。可以使用COPY命令重试批量加载。

数据提取功能入门

对于机器学习和业务分析用例成功而言，将所有数据放入数据湖至关重要，这对每个组织而言都是一项艰巨的任务。我们很高兴介绍Auto Loader和合作伙伴集成功能，以帮助我们成千上万的用户在构建高效数据湖的过程中获得帮助。这些功能现在可以作为预览。我们的文档提供了有关如何开始合作伙伴集成（Azure | AWS），自动加载程序（Azure | AWS）和复制命令（Azure | AWS）的更多信息，以开始将数据加载到Delta Lake。

閱讀更多 發家致富全靠它 的文章

關鍵字: 摄取 数据库 DevOps