这是我们Sellpoints的朋友的一个博客。Saman是Sellpoints的数据工程师，他负责开发、优化和维护我们的后端ETL。克里斯是Sellpoints的数据科学家，负责创建和维护分析管道，并开发用户行为模型。描述卖点最简单的方法是我们帮助品牌和零售商使用数据分析将购物者转化为购买者。Sellpoints的两个主要业务领域包括一些可以想象的最大和最复杂的数据集；我们为最大的在线零售商构建和联合交互式内容，以帮助品牌提高转化率，我们还为客户提供先进的展示广告解决方案。我们研究用户浏览零售网站、与我们的小部件交互或看到目标显示广告时的行为趋势。跟踪这些事件平均每秒有5500个独特的数据点。数据团队的主要任务是将我们的原始事件跟踪数据转化为聚合的用户行为信息，这些信息对于客户经理和客户的决策非常有用。我们跟踪人们在购买之前是如何与零售商网站互动的。我们单独的日志行表示客户与产品相关的一些内容之间的单一交互，然后我们将这些内容聚合起来，以回答诸如"一个人在购买前访问了多少次"或"人们在关闭窗口之前观看了多少演示视频，以及客户是否更有可能购买这些视频？"看了之后？"Sellpoints的数据团队负责创建用户行为模型，为我们跟踪的内容生成分析报告和可视化，并维护可视化基础设施。在本博客中，我们将描述Sellpoints如何不仅能够使用Databricks平台实现整个后端ETL，而且能够统一整个ETL。用例：集中式ETLSellpoints需要一个大数据处理平台，这个平台能够复制我们现有的基于Amazon Web Services（AWS）的ETL，但可以提高速度并有可能集成S3以外的数据源，包括MySQL和FTP。我们还希望能够测试MLlib，apachespark的机器学习库。我们非常高兴被接纳为Databricks的早期客户之一，并概述我们的目标：短期目标：在SparkSQL中复制现有的基于配置单元的ETL–将原始CSV日志文件处理为可用的表，优化文件大小，并根据需要进行分区将数据聚合到表中以进行可视化用Tableau提取可视化表格长期目标：在Scala/Spark中重写ETL并优化速度通过JDBC集成MySQL测试并生产MLlib算法将数据块进一步集成到我们的堆栈中我们在Sellpoints有一个小型数据团队，由三名数据工程师、一名数据科学家和两名分析师组成。因为我们的团队很小，而且我们自己没有DevOps资源来维护大型集群，所以Databricks是完美的解决方案。他们的平台允许分析师只需点击一个按钮就可以启动集群，而不必处理安装软件包或特定版本的软件。此外，Databricks为Spark提供了一个笔记本电脑环境，使数据发现非常直观。其易用性无与伦比，允许整个数据团队的用户调查我们的原始数据。架构和技术细节我们的直接目标是在SparkSQL中复制我们现有的基于Hive的ETL，结果很简单。SparkSQL可以直接实现配置单元库，并使用几乎完全相同的语法，因此传输现有的ETL非常简单，只需复制/粘贴并上传配置单元CSV Serde。不久之后，Databricks发布了他们的原生Spark CSV Serde，我们毫无疑问地实现了它。我们还需要将这些表提取到Tableau中，Databricks再次简化了这一点。Databricks默认为JDBC调用实现一个Thrift服务器，Tableau有一个SparkSQL连接器，它利用JDBC。我们需要修改一个设置：spark.sql.thriftServer.incrementalCollect=真这告诉Thrift服务器，数据应该以小增量发送，而不是将所有数据收集到驱动程序节点中，然后再将其推出，这将导致驱动程序在提取大量数据时很快耗尽空间。这样，在数据块中复制ETL就完成了。我们现在利用Databricks处理每天100GB的原始CSV数据。我们接下来的步骤包括学习Scala/Spark以及在Spark中可以进行的各种优化，同时也开始集成其他数据源。我们有一个SQL查询需要2个多小时才能完成，但是输出被保存到S3，供Databricks使用。查询花费了很长时间，因为它需要将11个表连接在一起，并构建一个包含10个字段的查找。虽然可以对MySQL数据库的结构或查询本身进行优化，但我们想，为什么不在Databricks中进行优化呢？我们能够将查询时间从2小时减少到不到2分钟；而且，由于SparkSQL语法包含了所有基本的SQL命令，所以实现查询就像复制/粘贴一样简单（注意，我们需要修改AWS设置才能使其工作）。以下代码的简化版本：进口com.mysql.jdbc.驱动程序//创建数据库连接val host="mysql数据库内部IP地址（10.x.x.x）"val port="3306"val user="用户名"val password="密码"val dbName="数据库名称"值dbURL=s"jdbc:mysql：//$host:$port/$dbName？user=$user&password=$password"//将表加载为valval表格=sqlContext.阅读.format（"jdbc"）.选项(地图("url"->s"$dbURL"，"数据库表"->"dbName.tabletooload","驱动程序"->"com.mysql.jdbc.Driver"））.加载（）//注册为临时SparkSQL表table.registerEmptable表（"表名"）//运行查询val表格数据=sqlContext.sql("""选择*来自表名""）我们也每天通过FTP接收和发送文件。我们需要下载并存储这些文件的副本，所以我们开始使用Databricks将它们下载到S3。这使我们能够进一步将ETL集中到数据块中。以下代码的简化版本：进口org.apache.commons网站.net.ftp.ftp客户端;进口org.apache.commons网站.net.ftp.ftp文件;进口org.apache.commons网站.net.ftp.FTPReply;进口java.io.File文件;进口java.io.FileOutputStream;val ftpURL="ftp.url.com"val user="用户名"val pswd="密码"val ftp=新的FTPClient（）ftp.connect（英尺/小时）ftp.enterLocalPassiveMode()ftp.登录（用户，pswd）val folderPath="路径下载自"val文件=ftp.list文件（s"/$folderPath"）.map（\getName）dbutils.fs.mkdirs（s"/$savePath"""）val outputFile=新文件"/dbfs/$savePath/文件名.txt""")val output=新文件输出流（outputFile）ftp.retrieveFile文件（s"/$文件夹/文件名.txt"，输出）输出.关闭()ftp.注销我们从将处理过的数据保存为CSV改为Parquet。Parquet是列式的，这意味着当您只使用数据中的某些列时，它可以忽略其他列。当你有数万亿行时，这有着巨大的速度提升，并允许我们减少等待初始结果的时间。我们尝试将单个parquet文件的大小保持在120MB左右，这是Spark的默认块大小，并允许集群快速加载数据（在使用小CSV文件时，这是一个瓶颈）。当我们实际分析数据时，将ETL步骤整合到一个单独的位置有更多的好处。我们总是试图保持在机器学习和离线分析技术的前沿，以识别新的用户群体，这就要求我们维护一组最新的可测试用户数据，同时能够将新技术的输出与我们当前使用的进行比较。能够将我们所有的数据整合到一个平台上大大加快了我们的迭代过程，特别是因为这个单一平台包括MLlib项目。益处和经验教训在实施Databricks后的1.5年中，我们能够：将我们的第一步ETL加快约4倍将mysql和ftp数据源直接集成到数据块中，并将这些部分的ETL速度提高100倍优化我们的分析处理数据在这个过程中，我们也学到了很多。一些提示和教训包括：将数据存储为拼花地板文件利用Spark的处理能力，通过JDBC在数据块中实现任何RDS ETL在保持较大的文件大小的同时尽可能地对数据进行粒度划分–加载数据很慢，但处理速度很快！要了解有关Sellpoints如何使用Databricks的更多信息，请阅读案例研究。特别感谢伊万·德贾诺维奇的编辑和耐心。免费试用Databricks。今天就开始吧