今天，我们很荣幸地宣布雪花和Databricks之间的合作伙伴关系，通过在雪花为云构建的数据仓库和Databricks的统一分析平台之间提供优化的生产级集成，帮助我们的客户进一步统一大数据和人工智能。在过去一年中，我们的联合客户，如Rue Gilt Groupe、Celtra和ShopRunner，要求我们两家公司加强整合和合作关系。这些客户和其他许多已经在一起使用我们的产品的客户分享了他们的使用案例和经验，并提供了惊人的反馈。虽然这两款产品都是同类产品中的佼佼者，而且都是基于云计算技术构建的，但我们的客户要求改进连接器的性能和可用性。为了回应我们共同客户的反馈，我们很高兴在Databricks运行时中引入优化的、生产就绪的Databricks雪花连接器。解决办法是什么？很简单，此连接器汇集了同类最佳的技术，因此您可以拥有业界领先的ETL、数据仓库和机器学习，而无需担心初始或正在进行的配置和设置。具体来说，Databricks和Snowflake现在提供了一个优化的内置连接器，允许客户使用Databricks无缝地读写数据。这种集成极大地改善了我们的客户的体验，这些客户以更少的设置更快地开始，并自动更新两种产品的改进。另外，雪花的自动查询下推可以将某些查询下推到雪花中。这就消除了在决定应该在哪里进行什么处理时的所有复杂性和猜测。使用优化的连接器，复杂的工作负载由Spark和Snowflake处理可以转换为SQL的工作负载。这可以提供性能和成本方面的好处，而无需任何手动操作或正在进行的配置。对于flash在线零售商Rue Gilt Groupe的架构师Stephen Harrison来说，这意味着"由于我们使用Snowflake作为主要数据源来访问关于我们的成员和产品的所有信息，[通过Databricks Snowflake connector]，直接连接到我们的数据仓库是无缝的，直接导入到Spark，无需任何耗时的ETL过程，并直接写回雪花。"现在让我们看看连接器的作用！使用数据块和雪花首先，您需要一个雪花帐户和一个Databricks帐户。一旦您登录到Databricks，请确保您已经在Databricks中创建了一个集群，使用Databricks运行时4.2或更高版本，并在Snowflake中创建了一个虚拟仓库（compute cluster）。同样值得再次检查您的Snowflake和Databricks帐户是否位于同一区域，以获得最佳性能和最低成本。您需要将您的雪花虚拟仓库设置为自动挂起和自动恢复，这样您只需为使用的内容付费。就这样！没有要加载的库和要管理的配置。我们已经在这个博客中简化了一些代码，但是您可以在这里跟随Databricks笔记本中的代码片段。有很多关于不同功能的细节，所以值得一看！配置连接首先我们需要配置连接。为此，我们将利用Databricks Secrets API安全地存储和加密用于访问雪花的凭据。val用户=dbutils.secrets.get（"数据仓库"，"雪花用户"）val密码=dbutils.secrets.get（"数据仓库"，"雪花密码"）一旦我们这样做，我们就可以设置读写这些数据的选项。val options=地图("sfUrl"->。雪花计算网/"，//将其替换为您自己的连接信息"sfUser"->用户，"sfPassword"->密码，"sfDatabase"->"演示"，"sfSchema"->"databricks_demo"，"sfWarehouse"->"演示群集")在Python中，它看起来像下面这样。#使用secrets API设置这些值用户=dbutils.secrets.get（"数据仓库"，"雪花用户"）密码=dbutils.secrets.get（"数据仓库"，"雪花密码"）#将其替换为您自己的连接信息选项={"sfUrl"：。雪花计算网/","sfUser"：用户，"sfPassword"：密码，"sfDatabase"："演示"，"sfSchema"："databricks_演示"，"sfWarehouse"："演示集群"，}把数据变成雪花将数据加载到雪花需要像加载其他数据源一样简单。没有库可加载或火花（或雪花连接器）版本担心-连接器是内置的！//在这个例子中，我们读取CSV数据值df=spark.read.format（"csv"）.option（"header"，"true"）.load（"/databricks datasets/adult/成人数据")//把数据写入雪花df.write.format格式（"雪花"）.选项（选项）.option（"dbtable"，"成人"）.mode（"append"）.save（）在上面的例子中，我们只做了一个简单的写。然而，许多客户利用Databricks对结构化和半结构化数据执行复杂的转换，以加载到雪花中，以用于下游分析或BI应用程序。使用Snowflake，您将获得本机JSON支持的额外好处，这意味着不需要对JSON数据进行转换。现在我们已经加载了数据，让我们以雪花的形式查询它。雪花查询数据加载数据后，很容易用雪花进行查询。启用雪花虚拟仓库后，只需打开雪花工作表并立即查询数据。下面是一个简单的查询，您可以运行它来操作数据：从成人限制5中选择*；在点击Run时，您将看到以下内容。随着数据现在加载到Snowflake中，业务分析师可以利用诸如SnowSQL之类的工具来查询数据，并针对这些数据运行许多商业智能应用程序。用户还可以利用雪花数据共享，以安全的方式与组织的其他部分或使用雪花的任何合作伙伴实时共享此数据。将Tableau或Looker等BI工具连接到您的Snowflake仓库也很容易，允许分析师查询存储在Snowflake中的大量数据。雪花是重要业务信息的优秀存储库，Databricks通过利用Databricks-Snowflake连接器将输入数据从Snowflake读入Databricks以进行模型训练，从而提供了在这些数据上训练机器学习模型所需的所有功能。训练机器学习模型为了训练机器学习模型，我们利用雪花连接器来提取雪花中存储的数据。您会注意到它与其他Spark数据源遵循相同的结构。因为它与Databricks运行时集成，所以它是零配置和生产准备。数据集=spark.read.format（"雪花"）\.选项（**选项）\.option（"dbtable"，"成人"）.load（）但是，有时出于性能或效率的考虑，可能需要限制从表中提取的数据。为此，请使用雪花连接器运行任意查询。例如，过滤到要训练ML算法的相关行。spark.read.format（"雪花"）\.选项（**选项）\.option（"query"，"从成人中选择工人阶级，婚姻状况，其中学历='Bachelors'）.load（）这是一个简单的示例，说明Databricks Snowflake连接器将自动将任何谓词甚至表达式下推到Snowflake中，这意味着您将获得开箱即用的最佳性能。预处理和特征生成现在我们已经加载了数据，我们可以开始在数据块中定义机器学习模型转换。例如，这里我们将定义一个管道，将分类变量转换为索引变量和一个热编码变量，以便输入到我们的机器学习算法中。从pyspark.ml导入管道从pyspark.ml.功能导入OneHotEncoderEstimator、StringIndexer、VectorAssemblercategoricalColumns=工人阶级"，"教育程度"，"婚姻状况"，"关系"，"种族"]numericCols=年龄"，"FNLWGT"，"教育"，"资本收益"，"资本损失"，"每周小时数"]assemblerInputs=[c+"classVec"表示分类列中的c]+numericCols所有列=分类列+数字列阶段=[]对于categoricalColumns中的categoricalColumns：stringIndexer=stringIndexer（inputCol=categoricalCol，outputCol=categoricalCol+"索引"）编码器=一个hotEncodeRestimator（inputCols=[stringIndexer.getOutputCol（）]，outputCols=[categoricalCol+"classVec"]）stages+=[stringIndexer，编码器]label_stringIdx=StringIndexer（inputCol="收入"，outputCol="label"）assembler=VectorAssembler（inputCols=assemblerInputs，outputCol="特性"）stages+=[label_stringIdx，汇编程序]我们现在有一个预处理的数据集来训练我们的机器学习算法。...阶段=管道管道模型=管道.fit（数据集）结果=pipelineModel.transform（数据集）selectedcols=label"，"features"]+所有列数据集=结果.选择（选定列）（培训数据，测试数据）=dataset.randomSplit（[0.7，0.3]，种子=100）#打印计数打印(培训数据.count())打印(测试数据.count())一旦我们完成了训练测试分割，我们现在可以使用交叉验证和随机森林分类器训练和评估我们的模型。从pyspark.ml.分类导入RandomForestClassifier从pyspark.ml.评估导入BinaryClassificationEvaluator从pyspark.ml.调整导入ParamGridBuilder，CrossValidatorparamGrid=（ParamGridBuilder（）.添加网格(射频最大深度，[2,4,6]）.添加网格(射频maxBins，[20，60]）.添加网格(射频数树，[5，20]）.build（））#生成交叉验证工具cv=交叉验证程序（estimator=rf，estimatorParamMaps=paramGrid，evaluator=evaluator，numFolds=5）cvModel=等速配合（培训数据）预测=cvModel.transform（测试数据）评估器。评估（预测）最后，我们可以保留我们的最佳模型，并用它进行预测。最佳型号=cvModel.bestModel公司最终预测=最佳模型转换（数据集）评估器。评估（最终预测）现在我们训练了这个模型并对其进行了评估