欢迎阅读本刊，了解 2021年9月 阿里云大数据产品 最新发布功能。大数据计算服务 - MaxCompute1、大数据计算服务 - MaxCompute 支持读取 OSS 上 Hudi、Delta Lake 格式文件 适用客户基于OSS 构建实时数据湖的客户，可以通过 MaxCompute 查询到实时数据，即时洞察业务数据变化。发布功能Delta Lake 和 Hudi是数据湖方案中常用的存储机制，为数据湖提供流处理、批处理能力。MaxCompute 基于阿里云 DLF、OSS产品提供了支持 Delta 或 Hudi 存储机制的湖仓一体架构。您可以通过 MaxCompute 查询到实时数据，即时洞察业务数据变化。查看文档 2、大数据计算服务 - MaxCompute 支持访问开启 Kerberos 认证授权机制的 Hadoop 集群 适用客户(1) 已经基于 Hadoop、阿里云 EMR 集群搭建数据平台，并开启 Kerberos 认证授权机制的客户；(2) 对集群资源弹性有需求，希望利用 MaxCompute 的弹性与大规模机器学习能力，快速迭代创新业务(3) 企业有多个 Hadoop 集群，需要各个集群之间数据打通，利用 DataWorks 构建企业数据中台发布功能大部分企业的 Hadoop 的生产集群开启了 Kerberos 认证授权机制，本次功能升级能够利用 MaxCompute 访问开启 Kerberos 认证授权机制的 Hadoop 集群，极大的扩展了湖仓一体对接 Hadoop 的适用范围。查看文档3、大数据计算服务 - MaxCompute 在华南1金融云（深圳）正式开服开放区域MaxCompute新增开通华南1金融云（深圳）区域，经过金融云认证的账号可以在产品售卖页按需开通申请链接DataWorks1、数据治理中心全新发布适用客户基于 DataWorks 和 MaxCompute 搭建数据平台，具备对数据存储、任务计算、代码开发、数据质量或数据安全等维度进行治理诉求的客户发布功能(1) 用于事前环节的数据治理问题预防(2) 用于事后环节的数据治理存量问题自动发现(3) 数据治理整体量化评估产品文档上海Rgion试用地址2、DataStudio支持用户通过扩展程序对页面功能进行卡点 适用客户使用DataWorks OpenAPI 以及扩展程序开发的用户发布功能DataStudio支持的扩展点包括文件发布、文件提交、文件删除产品文档3、数据服务API资产全面纳入数据地图管理范畴 适用客户已经基于DataWorks和各类数据源搭建数据平台，日常有数据接口API的开发工作，并需要进一步对各类数据源下的数据接口API进行全局查找、盘点和纳管的客户。发布功能数据服务API资产（向导、脚本、注册等类型API）已接入数据地图，支持企业场景和全局范围的数据接口查找及管理，包括全局API检索、热门API统计、独立API详情页、各数据源下的API分布等产品功能。产品文档1产品文档2交互式分析（Hologres）1、增强企业级运维能力，透出慢查询日志，帮助用户快速定位SQL问题 适用客户互联网行业（电商、教育、游戏等），有大规模实时数仓建设、交互式分析大屏、智能营销、风控等场景需求的客户；金融行业（证券、保险、银行等），有海量数据精细化分析、基于MaxCompute构建全栈数据仓库的客户； 能源、交通行业，有海量数据实时分析需求的客户；政务行业，有数据集中化、细粒度权限管控、平台赋能需求的客户；已搭建MaxCompute集中数仓，有交互式分析需求的客户。发布功能慢查询日志用来记录在 Hologres 中执行时间超过指定时间的查询语句。通过慢查询日志，可以查找出哪些查询语句的执行效率低，以便进行优化。慢查询日志是排查问题的 SQL 语句，以及检查当前 Hologres 性能的一个重要功能。从Hologres V0.10版本开始，系统提供了慢查询日志。慢Query的查询与分析可以帮助您对系统中发生的慢Query或失败Query进行诊断、分析和采取优化措施。产品文档1产品文档22、优化列存默认采用AliORC压缩格式，存储压缩比提高30%~50% 适用客户互联网行业（电商、教育、游戏等），有大规模实时数仓建设、交互式分析大屏、智能营销、风控等场景需求的客户； 金融行业（证券、保险、银行等），有海量数据精细化分析，基于MaxCompute构建全栈数据仓库的客户； 能源、交通行业，有海量数据实时分析需求的客户； 政务行业，有数据集中化，细粒度权限管控，平台赋能需求的客户；已经搭建MaxCompute集中数仓，有交互式分析需求的客户。发布功能从Hologres V0.10版本开始，行存表的存储格式全面升级为AliORC，存储压缩比提高30%~50%。产品文档

本期月刊将为您带来 MaxCompute 9月产品最新动态。导读9月产品发布最新动态9月技术直播回顾9月产品发布最新动态1、MaxCompute 产品在华南1金融云（深圳）正式开服 申请金融云认证>>开放区域MaxCompute新增开通华南1金融云（深圳）区域，经过金融云认证的账号可以在产品售卖页按需开通。2、MaxCompute 支持读取 OSS 上 Hudi、Delta Lake 格式文件 查案文档 >>适用客户基于OSS 构建实时数据湖的客户，可以通过 MaxCompute 查询到实时数据，即时洞察业务数据变化。发布功能Delta Lake 和 Hudi是数据湖方案中常用的存储机制，为数据湖提供流处理、批处理能力。MaxCompute 基于阿里云 DLF、OSS产品提供了支持 Delta 或 Hudi 存储机制的湖仓一体架构。您可以通过 MaxCompute 查询到实时数据，即时洞察业务数据变化。3、MaxCompute 支持访问开启 Kerberos 认证授权机制的 Hadoop 集群 适用客户已经基于 Hadoop、阿里云 EMR 集群搭建数据平台，并开启 Kerberos 认证授权机制的客户；对集群资源弹性有需求，希望利用 MaxCompute 的弹性与大规模机器学习能力，快速迭代创新业务的客户；企业有多个 Hadoop 集群，需要各个集群之间数据打通，利用 DataWorks 构建企业数据中台的客户。发布功能大部分企业的 Hadoop 的生产集群开启了 Kerberos 认证授权机制，本次功能升级能够利用 MaxCompute 访问开启 Kerberos 认证授权机制的 Hadoop 集群，极大的扩展了湖仓一体对接 Hadoop 的适用范围。9月技术直播回顾「 第一讲 」基于MaxCompute的智能推荐解决方案   直播回放 >>  阅读文字版 >>「 第二讲 」基于MaxComput+开放搜索的电商、零售行业搜索开发实践   直播回放 >>  阅读文字版 >>「 第三讲 」基于MaxCompute+PAI的用户增长方案实践   直播回放 >>  阅读文字版 >>「 第四讲 」基于MaxCompute的实时数据处理实践   直播回放 >>  阅读文字版 >>「 第五讲 」基于MaxCompute分布式Python能力的大规模数据科学分析 直播回放 >>  阅读文字版 >>「 第六讲 」基于MaxCompute +Hologres 的人群圈选和数据服务实践 直播回放 >>  阅读文字版 >>「 第七讲 」基于 MaxCompute SQL 的半结构化数据处理实践 直播回放 >>  阅读文字版 >>「 第八讲 」基于Delta lake、Hudi格式的湖仓一体方案 直播回放 >>  阅读文字版 >>

此文是对 Spark on MR3 资料的翻译原文链接：https://www.datamonad.com/post/2021-08-18-spark-mr3/代码链接：https://github.com/mr3project/spark-mr3MR3 是一个通用的执行引擎，原生支持 Hadoop 和 Kubernetes。虽然 Hive on MR3 是主要应用，但 MR3 也可以轻松执行 MapReduce/Tez 作业。例如，Hive on MR3 可以不依赖 MapReduce 进行压缩，因为执行压缩的 MapReduce 作业会被直接发送到 MR3。随着 MR3 1.3 的发布，开发团队引入了另一个主要应用 – Spark on MR3。简而言之，就是使用 MR3 作为执行后端的 Apache Spark。MR3 上的 Spark 是作为 Spark 的附加组件实现的，它利用 MR3 来实现 Spark 的调度后端。因此，在 MR3 上运行 Spark 不需要对 Spark 进行任何更改，并且用户在 MR3 上安装 Spark 时可以使用 Spark Binary Distribution。在详细介绍 Spark on MR3 之前，应该回答一个关键问题——为什么我们需要一个替代的 Spark 执行后端？ 毕竟 Spark 是一个成熟的系统，已经为 Hadoop 和 Kubernetes 提供了原生支持，所以 Spark on MR3 似乎没有带来额外的优势。然而，Spark on MR3 不仅仅是为了展示 MR3 作为执行引擎的能力。相反，它解决了 Apache Spark 的一个重要架构限制。动机在 MR3 上开发 Spark 的主要动机是允许多个 Spark 应用程序共享计算资源，例如 Yarn 容器或 Kubernetes Pod。 对于普通 Spark，不同的 Spark 应用程序必须维护自己的一组 executor，因为 Spark 缺乏在 Spark 应用程序之间回收计算资源的功能。相反，Spark 依靠资源管理器（例如 Yarn 和 Kubernetes）来分配计算资源。请注意，Apache Livy 不是这里的解决方案，因为它只允许多个用户通过 REST API 共享 Spark 应用程序。对于 Spark on MR3，MR3 DAGAppMaster 的单实例管理 Spark 应用程序之间共享的所有计算资源。下图说明了在 Kubernetes 上运行的 Spark on MR3，其中四个 Spark driver 共享一个公共 DAGAppMaster Pod。当 Spark driver 终止时，其执行程序 Pod 不会立即释放回 Kubernetes。相反，DAGAppMaster 尝试通过将空闲 Pod 重新分配给那些请求更多计算资源的 Spark driver 或将它们保留在备用池中以备将来使用。因此，当多个 Spark 应用程序并发运行时，Spark on MR3 减少了获取和释放计算资源的开销。由于 MR3 使用 Java 虚拟机(JVM)，回收计算资源也转化为减少 JVM 预热开销。此外，Spark on MR3 将自身与资源管理器的调度策略分离。例如，无论资源管理器的调度策略如何，我们都可以在 Spark 应用程序之间强制执行公平调度。Spark on MR3 在频繁创建和销毁 Spark 应用程序的云环境中特别有用。在云环境中，配置计算资源的开销远非可以忽略不计。例如，在 Amazon EKS 上预置新的 Kubernetes 节点可能需要几分钟时间。通过允许 DAGAppMaster 保留计算资源的备用池，Spark on MR3 可以有效地最小化计算资源的开销。因为 Spark 作业在提交后立即执行，它可以实现云成本的降低以及用户体验的改进。应用Spark on MR3 是作为 Spark 的附加组件实现的。Spark-MR3 作为外部集群管理器，提供 Spark 调度后端的实现。Spark-MR3 嵌入在 Spark driver中，并在 Spark driver 和 MR3 之间传输消息。在内部，它提供了 Spark TaskScheduler 和 SchedulerBackend 的实现，并创建了 MR3Client 以便与 MR3 进行通信。当 Spark DAGScheduler 想要执行新的 Spark stage 时，Spark-MR3 将 stage 的格式转换为 DAG，然后发送到 MR3 DAGAppMaster。DAG 由运行在 ContainerWorkers 内的 Spark executor 执行，并将结果发送回 Spark。由于 Spark-MR3 将所有来自 Spark 的请求发送到 MR3，因此从 Spark 的角度来看，MR3 是类似于 Yarn 和 Kubernetes 的资源管理器。这是有效的，因为 Spark 与底层集群管理器无关，而 MR3 能够在与真正的资源管理器通信后提供计算资源。由于 Spark 使用 Java ServiceLoader 机制加载外部集群管理器，因此在 MR3 上运行 Spark 不需要对 Spark 进行任何更改。实验现在开发团队展示实验结果来证明 Spark on MR3 的性能。除了任务调度和资源管理，Spark on MR3 的行为方式与普通 Spark 相同。首先，开发团队展示了 Spark on MR3 在单个 Spark 应用程序上与普通 Spark 表现一致。接下来，开发团队将展示当多个 Spark 应用程序同时运行时，Spark on MR3 的运行时间大幅减少。在实验中，开发团队使用两个运行 HDP（Hortonworks Data Platform）3.1.4 的集群。每个工作节点运行一个 ContainerWorker，或者说是一个 Spark executor。Gold 集群，由 2 个主节点、34 个工作节点（每个节点具有 24 个内核和 96 GB 内存）和十千兆网络组成。Grey 集群，1 个主节点，9 个工作节点（每个节点有 12 个内核和 48 GB 内存）和一千兆网络。慢网络使得 Grey 集群适合评估普通 Spark 和 Spark on MR3 中延迟调度的有效性。开发团队使用 Spark 多用户基准测试作为工作负载。Spark 应用程序中的每个线程都提交相同的 8 个 TPC-DS 基准查询序列（#19、#42、#52、#55、#63、#68、#73、#98）。开发团队使用的比例因子为 100（大约相当于 100GB），它可以保持每个工作节点处于忙碌状态，也可以清楚地突出回收 ContainerWorkers 的效果。开发团队以文本格式存储数据集而不进行压缩。单个 Spark 应用程序Spark on MR3 提供了两种任务调度策略——FIFO 调度和公平调度，它们规定如何在同一个 Spark 应用程序中分配 Spark 作业。Spark on MR3 也以类似于普通 Spark 的方式实现延迟调度。对于单个 Spark 应用程序的测试，开发团队使用 1 秒延迟进行延迟调度（例如，设置普通 Spark 的参数 spark.locality.wait 为 1 秒 ）。下图显示了在 Gold 集群中使用普通 Spark 和 Spark on MR3 运行单个 Spark 应用程序的结果。Spark driver 创建八个线程，每个线程提交相同的八个 TPC-DS 查询序列。（因此它总共提交了 $8 \times 8 = 64$ 个 Spark 作业。）开发团队观察到，在 FIFO 调度下，Spark on MR3 的运行速度与普通 Spark 一样快，但在公平调度下，速度稍慢。Grey 集群测试结果完全不同。开发团队观察到，Spark on MR3 比普通 Spark 执行 Spark 作业的完成速度要快得多，FIFO 调度为 3034 s 与 8042 s，公平调度为 3583 s 与 8017 s。运行时间的巨大差异是由于任务调度的不同，特别是延迟调度在普通 Spark 和 Spark on MR3 中是如何工作的。下图显示了没有匹配机器的 Spark 任务的百分比。例如，一个带有 grey1 和 grey2 位置标签的 Spark 任务，如果在 grey3 上执行，则没有匹配到机器上。开发团队观察到，普通 Spark 的百分比要高得多，FIFO 调度为 1.8% 与 22.7%，公平调度为 7.2% 与 26.8%。结合使用慢速（1 Gb）网络，高百分比会产生许多落后的任务，从而显着增加运行时间。多个 Spark 应用程序独立于任务调度策略，Spark on MR3 提供了两个策略——FIFO 调度和公平调度，指定如何在多个 Spark 应用程序之间分配 ContainerWorkers（即计算资源）。为了测试多个应用程序，开发团队禁用延迟调度。下图显示了在使用 Yarn 公平调度的 Gold 集群中使用普通 Spark 运行 12 个相同的 Spark 应用程序的结果。竖条显示在相应工作节点上运行的 Spark executor 的进度，其颜色表示拥有 Spark executor 的 Spark 应用程序。可以观察到，尽管使用了公平调度，但 Spark 应用程序在 Yarn 分配的计算资源总量方面存在很大差异。我们还看到竖线之间的许多间隙，它们表示 Spark executor 终止后和另一个 Spark executor 启动之前的空闲时间。下图显示了在 Gold 集群中使用公平调度在 Spark on MR3 运行相同的 12 个 Spark 应用程序的结果。 顺便说一句，资源管理器本身的调度策略对实验来说并不重要。可以观察到，Spark on MR3 通过回收 ContainerWorkers 快速实现了 Spark executor 之间工作节点的公平分配。 请注意，Spark on MR3 不会在垂直条之间产生间隙，因为在整个实验过程中，每个工作节点上都会回收一个 ContainerWorker。 在我们的实验中，与普通 Spark 相比，Spark on MR3 将运行时间减少了 25.6%（从 907.7s 到 674.6s）。下图显示了当 Yarn 和 Spark on MR3 都使用 FIFO 调度时 Gold 集群中的结果。开发团队以 60 秒的间隔提交每个 Spark 应用程序。普通 Spark 和 Spark on MR3 都一次运行一个 Spark 应用程序，并按照提交的顺序完成所有 Spark 应用程序。 请注意，一旦第一个 Spark 应用程序终止，Spark on MR3 就会更快地完成剩余的 Spark 应用程序，因为 JVM 已经预热。 在实验中，与普通 Spark 相比，Spark on MR3 将运行时间减少了 36.3%（从 797.3s 到 1144.5s）。总结Spark on MR3 易于使用在 Hadoop 和 Kubernetes 上，因为它与 Apache Spark 的不同之处仅在于任务调度和资源管理。 除了在 Spark 应用程序中回收 ContainerWorkers 之外，用户还可以利用 MR3 的容错、推测执行和自动缩放等特性。

我是一名由电气工程跨考到软件工程的研二学生，在最近的实习中，在对Hadoop的操作中，使用到了Linux，在实践里认识到了Linux的重要性，开始了Linux的学习，在视频课中，了解到了阿里云“飞天加速计划”对高校生的支持，使我萌生了申请云服务器进行Linux操作学习的想法，两周的学习时间宝贵并且短暂，在此对目前的学习进行简单总结，并对后续基于阿里云服务器的学习实验进行规划。目前主要进行大数据相关的工作和研究，这离不开Hadoop集群，之前自己尝试使用虚拟机搭建Hadoop集群，遇到了很多困难，而且虚拟机占用内存大，使用不方便，最近实习时，每天使用公司搭建在docker容器中，部署在服务器上的Hadoop集群，通过编写shell脚本，使用xshell连接服务器，xftp上传本地的shell脚本，在Hadoop集群上运行脚本，完成海量数据的清理和融合。工作中，对最新技术的使用，让我感受到Hadoop，云计算，docker之于大数据工作和研究的意义，而xshell，xftp，阿里云服务器对于在校学习者的友好也让我有了机会在实习工作之余，有机会搭建自己的服务器和集群，以便学习实验使用。最后，对后续基于阿里云服务器的学习和实验做初步的规划1.利用服务器打好Linux基础2.在服务器上安装docker，在docker容器中部署Mysql，Redis，jupyter等工具3.尝试结合docker部署hadoop集群在服务器上4.进行爬虫练习，将爬取数据通过Hadoop集群，完成数据的存储，清洗和转换预处理，借此过程熟悉Hadoop集群的使用，并为后续实验做准备5.利用云服务器的计算资源，使用Hadoop集群将完成预处理的数据导出，进行实验。

阿里云ECS云服务器分为多种:通用型、计算型、高主频型、共享型、异构GPU型及裸金属服务器等规格，ECS云服务器选择实际使用场景说明： ECS云服务器实例规格使用场景对照 阿里云ECS云服务器实例规格可以分为以下多种规格： x86计算：通用型、计算型、内存型、大数据型、本地SSD、高主频及共享型等； 异构计算GPU/FPGA/NPU：GPU计算型、GPU虚拟化型、FPGA计算型、NPU计算型； 弹性裸金属服务器（神龙）：CPU型、GPU型； 超级计算集群：CPU型、GPU型。 可以参考下方实际应用场景选择对应的ECS实例规格： 根据实际应用场景选择对应的ECS实例规格，关于ECS云服务器实例规格CPU、内存、网络性能等参数，可以参考官方文档：ECS云服务器实例规格CPU内存网络等性能参数详解 通用型 通用型 通用型 通用型 通用型 通用型 异构计算 异构计算 裸金属/高性能计算 通用型g6/g5、通用网络增强sn2ne 计算型c6/c5、计算网络增强型sn1ne 内存型r6/r5、内存增强型re6、内存增强型se1ne、内存型se1 高主频hfc6/hfg6/hfr6/hfc5/hfg5 本地SSD l2/l2g/l2ne/l2gne/l1 大数据型 d2s/d1ne/d1 GPU计算型gn6i/gn6v/gn6e/gn5、轻量级GPU计算型vgn6i FPGA计算型f3/f1 弹性裸金属ebm、超级计算集群scc、弹性高性能计算e-hpc 中小型数据库 数据处理任务 企业后台应用 高性能网站前端节点 Web服务器 批量计算 分布式分析 对战类游戏 高性能科学 工程类应用 平台广告 视频编解码 高性能数据库 数据挖掘和分析 Redis、Memcached等缓存 内存型数据库 高性能科学计算 高性能前端节点 关系型数据库 NoSQL数据库 数据仓库 内存型数据库 Hadoop/Spark集群（实时） MapReduce分布式计算Hadoop/Spark 分布式文件系统日志和数据处理 AI深度学习 视频处理 图形可视化 科学计算 基因组学研究视频编解码 图像转码 金融分析 高性能数据库 高性能网站前端节点 数据处理任务 企业后台应用 高性能计算 科学计算 正式购买之前，推荐先领取代金券在购买（附代金券领取地址：阿里云官方云小站），可帮助我们进一步降低购买成本。

大家好，我是冰河~~今天跟大家聊聊如何使用Java7提供的Fork/Join框架实现高并发程序。好了，开始今天的主题吧！Fork/Join框架位于J.U.C(java.util.concurrent)中，是Java7中提供的用于执行并行任务的框架，其可以将大任务分割成若干个小任务，最终汇总每个小任务的结果后得到最终结果。基本思想和Hadoop的MapReduce思想类似。 主要采用的是工作窃取算法（某个线程从其他队列里窃取任务来执行），并行分治计算中的一种Work-stealing策略为什么需要使用工作窃取算法呢？假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。工作窃取算法的优点充分利用线程进行并行计算，并减少了线程间的竞争工作窃取算法的缺点在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且该算法会消耗更多的系统资源，比如创建多个线程和多个双端队列。Fork/Join框架局限性对于Fork/Join框架而言，当一个任务正在等待它使用Join操作创建的子任务结束时，执行这个任务的工作线程查找其他未被执行的任务，并开始执行这些未被执行的任务，通过这种方式，线程充分利用它们的运行时间来提高应用程序的性能。为了实现这个目标，Fork/Join框架执行的任务有一些局限性，如下所示。任务只能使用Fork和Join操作来进行同步机制，如果使用了其他同步机制，则在同步操作时，工作线程就不能执行其他任务了。比如，在Fork/Join框架中，使任务进行了睡眠，那么，在睡眠期间内，正在执行这个任务的工作线程将不会执行其他任务了。在Fork/Join框架中，所拆分的任务不应该去执行IO操作，比如：读写数据文件任务不能抛出检查异常，必须通过必要的代码来出来这些异常Fork/Join框架的核心类Fork/Join框架的核心是两个类：ForkJoinPool和ForkJoinTask。ForkJoinPool负责实现工作窃取算法、管理工作线程、提供关于任务的状态以及执行信息。ForkJoinTask主要提供在任务中执行Fork和Join操作的机制。示例代码示例代码如下：package io.binghe.concurrency.example.aqs; import lombok.extern.slf4j.Slf4j; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.Future; import java.util.concurrent.RecursiveTask; @Slf4j public class ForkJoinTaskExample extends RecursiveTask<Integer> { public static final int threshold = 2; private int start; private int end; public ForkJoinTaskExample(int start, int end) { this.start = start; this.end = end; } @Override protected Integer compute() { int sum = 0; //如果任务足够小就计算任务 boolean canCompute = (end - start) <= threshold; if (canCompute) { for (int i = start; i <= end; i++) { sum += i; } } else { // 如果任务大于阈值，就分裂成两个子任务计算 int middle = (start + end) / 2; ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle); ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end); // 执行子任务 leftTask.fork(); rightTask.fork(); // 等待任务执行结束合并其结果 int leftResult = leftTask.join(); int rightResult = rightTask.join(); // 合并子任务 sum = leftResult + rightResult; } return sum; } public static void main(String[] args) { ForkJoinPool forkjoinPool = new ForkJoinPool(); //生成一个计算任务，计算1+2+3+4 ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100); //执行一个任务 Future<Integer> result = forkjoinPool.submit(task); try { log.info("result:{}", result.get()); } catch (Exception e) { log.error("exception", e); } } }推荐阅读《实践出真知：全网最强秒杀系统架构解密，不是所有的秒杀都是秒杀！！》《注意：线程的执行顺序与你想的并不一样！！》《深入解析Callable接口 》《两种异步模型与深度解析Future接口！》《解密SimpleDateFormat类的线程安全问题和六种解决方案！》《不得不说的线程池与ThreadPoolExecutor类浅析》《深度解析线程池中那些重要的顶层接口和抽象类》《从源码角度分析创建线程池究竟有哪些方式》《通过源码深度解析ThreadPoolExecutor类是如何保证线程池正确运行的》《通过ThreadPoolExecutor类的源码深度解析线程池执行任务的核心流程》《通过源码深度分析线程池中Worker线程的执行流程》《从源码角度深度解析线程池是如何实现优雅退出的》《ScheduledThreadPoolExecutor与Timer的区别和简单示例》《深度解析ScheduledThreadPoolExecutor类的源代码》《由InterruptedException异常引发的思考》《浅谈AQS中的CountDownLatch、Semaphore与CyclicBarrier》《浅谈AQS中的ReentrantLock、ReentrantReadWriteLock、StampedLock与Condition》《朋友去面试竟然栽在了Thread类的源码上》好了，今天就到这儿吧，我是冰河，我们下期见~~

真不知道该怎么开头，一转眼就到大四找实习的时候，现在也没有什么课程和作业了，每天就是在寝室里找合适的单位，然后写简历，投简历，等面试。时间还是比较空余，于是就想着做一个自己的网站，阿里云服务器就是我建站的首选。我第一次接触阿里云服务器是大三下学期，也是在一次选修课上知道了现在很多厂商选择在云服务器上建网站，下了课我就搜索云服务器相关知识。其实我很早就对云计算有所耳闻，只是经过那次深入了解才知道云服务器的运用竟然会如此广泛，从而也萌生了租一个自己的云服务器的想法。 我们学校没有相关的课程，也没有教学账号，我只好自己去找相关的教程。最后在服务器上配置了一些环境，跑了一些程序，然后慢慢的因为学业和一些琐事而把服务器荒废了，现在想想还是觉得有点浪费，如果能当时能坚持维护好自己的服务器，现在我应该有一个内容充实的博客了吧。这学期由于我的时间有开始充裕起来，我决定重启服务器，把以前搭建的环境都扔掉，从零开始，尽量把它做好。正好我又申请到了飞天加速计划，现在就有了2台云服务器，于是就想先试试分布式存储。下面是我在云服务器上安装Hadoop环境的大概过程，配置Hadoop集群安装的过程暂时没有保留，等有时间再更新的过程：1、建立一个新用户Hadoop，并赋予管理员权限2、更新apt3、安装ssh，设置ssh无密码登录登录ssh每次打开都需要输入密码，有点麻烦，所以设置免密码登录（输入ssh-keygen -t rsa 然后一直按Enter）然后执行以下语句cd .sshcat id_rsa.pub >> authorized_keys4、下载Java安装包tar -zxvf openjdk-8u41-b04-linux-x64-14_jan_2020.tar.gz5、解压Java安装包6、移动并重命名JDK 查看解压后jvm修改文件名称为Java87、配置Java环境进入配置文件cd ~vim ~/.bashrc在文件头输入以下配置信息export JAVA_HOME=/usr/lib/jvm/java8export JRE_HOME=${JAVA_HOME}/jreexport CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/libexport PATH=${JAVA_HOME}/bin:$PATH最后记得 source ~/.bashrc7、查看是否配置成功8、下载Hadoop安装包镜像文件可以上https://mirrors.tuna.tsinghua.edu.cn/apache/hadoop/common/ 查看解压配置环境最后看一下版本安装教程暂时完了，接下来就是集群安装配置了这里我要对所有和我一样的初学者说一句，一定要学会使用Linux的命令行，尽量不要使用图形化界面，这会严重影响服务器运行，虽然放弃了图形化界面会有不适应，但是很多企业服务器都是不使用图形界面的，之前我看一个介绍Dell的一款服务器的视频，那款最近推出的服务器居然还在用VGA接口，据说因为他们是集成管理，也用不上图形界面，安装HDMI接口也是浪费。其实我也是一个初学者，对云计算还不算很了解，但我的一位前辈和我说学习这些就是要持之以恒，因为技术是在不断发展的，一个在IT行业里耕耘10年的开发者在面对新技术的时候其实是和我们站在一条起跑线上的，所以我们也不要觉得现在才开始学习太晚了，最好的学习时间永远是今天。我对云服务也是很看好，从技术角度来说，现在的网络条件是十分良好，虚拟技术也已经十分成熟；从市场角度来看，现在各大厂商都在争夺云服务器这块宝地，说明这是一个很有前景的行业。我们要做的就是不断精进武艺，尽量跟上时代的发展。

本人就读的专业是数据科学与大数据技术，目前大三。大二时学习了Hadoop、Hive、Hbase等一些需要在服务器集群上运行的大数据工具，在校学习时使用的是本地的三台Linux虚拟机搭建的集群。我一直想在真正的服务器上搭建一个hadoop集群，经同学介绍，了解了“飞天加速计划·高校学生在家实践”活动。 在使用服务器这段时间，我的操作系统是centOS7，我在服务器上安装了java8、python3.7和mysql，部署了一个自己写的密码管理系统的javaweb项目（本来是想把服务器和两台本地虚拟机共同搭建Hadoop集群，但尝试后发现服务器无法ping通虚拟机，终没能解决）。 在这期间，我找到了一些能够提高服务器使用效率的工具和方法。首先是一个我认为比较好用的远程控制终端的软件——finalshell。我以前一直在用xshell，但对于服务器的性能监控不了解，而在finalshell中，我可以直观的看到服务器的CPU占用、内存占用、网络等实时状况，也可以以树状图的形式看到文件分布。对于服务器的整体情况掌握有很大帮助。 其次是Linux上的会话控制软件——tmux。这个软件可以进行终端复用，也就是可以在一个终端窗口中开启多个会话。虽然在终端控制软件上双击即可开启一个新的窗口，但是如果直接操作服务器的话还是要用tmux。tmux还有一个我很喜欢的功能，就是可以在自己建的会话中看到以前执行的命令。在终端控制软件中如果关闭窗口再重新打开，你的历史命令就看不到了，而在tmux中可以复现。 再次是Mycli这款mysql辅助软件。它可以想idea一样提示mysql中的命令，以免我偶尔手残敲错。最后是大名鼎鼎的vim，它扩展了原本vi中没有的功能，配置各种软件的参数方便了许多。 在使用期间，我也遇到了一些问题。我在使用rm -rf命令时偶尔会一不小心手残删错，每次都要吓死。我以前也看过公司员工在生产环境中误删数据库的新闻。我想到的解决办法是仿照Windows上的回收站操作。我可以新建一个名为RecycleBin的文件夹作为回收站，写脚本将删除操作替换为将被删除的文件mv到RecycleBin文件夹下，然后对存放时间超过七天的文件使用定时脚本进行统一删除。后来我了解到也可以使用Linux的自定义命令实现。 在“高校学生在家实践”的活动中，我学到了许多东西。我体验到了真正的服务器与本地虚拟机之间的差异，真正的服务器需要你时刻关注其硬件使用状况，而本地虚拟机由于配置较高，就没有考虑过这些。我还感受到了服务器的延迟问题，我在使用本地Navacat与服务器上的mysql进行连接与数据读取写入时，有时需要加载好一会。这期间最让我难忘的是我写javaweb项目过程。这学期我有javaweb选修课，恰好也有服务器，所以我想在服务器上部署一个javaweb项目。但是服务器只有两周时间，而且我只有Java基础，所以我必须提前学习javaweb的相关知识。当时老师才刚刚讲到DAO层。于是我花了好几天把maven、springboot、mybatis、html、css、thymeleaf等的入门知识都自学了一遍,自己完成了Service层和Controller层还有前端的编写。在这期间遇到各种各样谜之错误，有的网上还找不到解决方法，只能自己翻源码，无数次想要放弃，但又觉得机会难得，不能白白浪费。最后功夫不负有心人，终于完成了一个十分简易的网站，但由于我的编写经验不足，网站的页面和逻辑方面的问题有很多，上线测了几把就匆匆下线。但是这个过程让我感到自己成长了许多，觉得以前看起来很高深的东西比如springboot，真正投入学习时其实也不是很难，只要肯坚持下去，总会有个名堂的。以后我也会继续我的大数据之路，如果能够成功免费续租服务器，我将会继续完善我的网站。

上篇文章已经成功集成了Trino和CarbonData，这篇文章主要是测试我们的功能和性能方面的一些测试，测试出CarbonData相对于社区其他的一些文件格式的优势以及验证集成性能和可靠性，包括是否能正确启动集群，正确识别Carbondata文件进行读取数据做count，聚合分析等建表语句：/* * create table by user pay list */CREATE TABLE IF NOT EXIST CARBONDATA_PAYMENT (    `gameId` int,    `unionId` int,    `orderId` STRING,    `userId` INT,    `createTime` bigint,    `payTime` bigint,    `status` int,    `money` int) stored as carbondata生成测试数据的Spark代码：package org.apache.carbondata.trinoimport org.apache.spark.SparkContextimport org.apache.spark.sql.CarbonSession._import org.apache.spark.sql.SparkSessionimport scala.util.Random/** * @Description Build Test Data Store AS CarbonData * @Author chenzhengyu * @Date 2021-09-10 20:37 */object TestDataFromCarbonData {  def main(args: Array[String]): Unit = {    val sc = new SparkContext();    val spark = SparkSession.builder().enableHiveSupport().getOrCreate()    val carbon = SparkSession.builder().config(sc.getConf).getOrCreateCarbonSession("hdfs://localhost:9000/user/hive/warehouse")    //build random num    val randomGameId = new Random(5);    val randomMoney = new Random();    import spark.implicits._    //build data    val dfTemp = spark.sparkContext.parallelize(1 to 1000000000).map(x => (randomGameId.nextInt(5), x % 8, "20210920" + (x % 100), x % 5, System.currentTimeMillis() + x % 500L, System.currentTimeMillis() + x % 500L, randomGameId.nextInt(4), randomMoney.nextInt(1000))).toDF("gameId", "unionId", "orderId", "userId", "createTime", "payTime", "status", "money")    //create temp view    dfTemp.createOrReplaceTempView("temp")    //create table    carbon.sql("CREATE TABLE CARBONDATA_PAYMENT (`gameId` int, `unionId` int, `orderId` STRING, `userId` INT, `createTime` bigint, `payTime` bigint,`status` int,`money` int ) stored as carbondata")    //show tables    carbon.sql("show tables").show    //insert data and select    spark.sql("insert into table CARBONDATA_PAYMENT select * from temp")    spark.sql("select count(1) from CARBONDATA_PAYMENT").show()    sc.stop()  }}这一部分代码我们可以使用Spark-Shell或者使用打包工具打包成JAR，通过Standalone模式或者YARN模式进行执行，执行的流程如下：Spark框架的核心是一个计算引擎，整体来说，它采用了标准 master-slave 的结构。如上图所示，它展示了一个 Spark执行时的基本结构。图形中的Driver表示master，负责管理整个集群中的作业任务调度。图形中的Executor 则是 slave，负责实际执行任务我一共插入了大概10亿条数据左右，平均每个分区插入2.5E条数据：生成的文件在HIVE目录下，观察如下：通过Trino的Client进行查询 SELCT COUNT(1) AS TOTAL FROM CARBONDATA_PAYMENT 我们一共生成了 10E条数据接下来我们做聚合查询，通过查询。我们通过模拟数据一共5个用户id，然后统计他们的总充值金额进行查询。SELECT USERID,SUM(MONEY) AS TOTAL FROM CARBONDATA_PAYMENT GROUP BY USERID结果如下图这个流程针上面我做一个简单的总结：首先我们通过Spark 集成 Carbondata成功后，去生成Carbondata文件，我们通过SparkSession 这个类去生成一个临时会话，然后通过Spark RDD 去模拟一些游戏玩家订单充值数据，然后注册为临时表Temp。通过在Spark上连接Hive元数据库创建Carondata表，名字为CARBONDATA_PAYMENT的表。然后通过INSERT语句把刚刚的生成的数据插入到Hive表上，这个过程大概在10分钟左右（受限于本机磁盘和CPU性能影响），后续我们通过观察即可发现Hadoop的HDFS上已经写入了对应的数据了。接着我们就可以关闭Spark了，开始启动Presto（Trino）的服务端和客户端，按照配置文件配置好Hive的MetaStore地址和Catlog，我们使用Trino客户端连接进去。连接进去后我们通过网页打开控制台进行观察SQL的执行情况。然后我们执行简单查询看看是否能正常工作，检测完毕后，我们执行聚合查询，模拟查询根据玩家ID的总充值金额。然后在不同数据量（1w->10w->100w->1000w->1e->10e）的查询下的执行情况。然后我们再模拟查询某个玩家订单号去查询相关的详情数据情况，并且最后对此进行总结。与此同时，我们模拟超出内存超出执行的情况，保护查询避免集群出现OOM的机制受限于本机性能发挥，基于刚刚按照导师要求进行的大规模数据测试结论如下：基本查询、聚合查询等查询基本无问题，证明新版本Presto（Trino）集成成功在做Count数据操作时候非常的快，基本1-2s可以出到结果，针对复杂的条件Count，如果有索引列也很快能查询出来通过观察HDFS文件路径，可以发现Index文件是单独存储的一份文件，这样做的目的是提升数据读取效率针对维度的查询明细数据，例如根据订单号进行查询，我们可以发现在10E数据中秒级返回查询的明细数据，这个速度是十分快的如果在执行时候，超过允许最大执行内存，则会集群有停止执行的机制，避免内存oom

本文作者 孟硕 阿里云智能 产品专家直播视频请点击 直播观看一、最佳实践背景整个最佳实践是基于MaxCompute的湖仓一体架构，模拟公司使用场景。比如公司 A 使用云上关系型数据库 RDS 作为自己的业务库，同时使用阿里云 EMR 系统做日志数据采集。将数据汇集到云上对象存储 OSS 上，引入了数据湖常会用的存储机制 Delta Lake 和 Hudi 为数据湖提供流处理、批处理能力。通过 MaxCompute 查询到实时数据，即时洞察业务数据变化。 整个场景demo的架构是，云上EMR产生的实时变化的数据，包括在线数据库RDS，通过数据入湖，然后实时的把数据变化体现在归档的OSS 上。同时MaxCompute跟其他引擎一起分析OSS上的数据。湖仓一体架构：异构数据平台融合因为企业内部会有很多业务线，不同的部门，因为本身业务的需求及员工的技术栈几个方面的原因，导致采用的数据架构不一样，数据平台也不一样。技术架构有Hadoop技术体系，也有云上全托管架构，所以造成不同的部门对技术架构，使用的技术平台不一样，也造成了数据割裂的情况。湖仓一体就是帮助企业把异构数据平台做一个打通，底层数据可以相互访问，中间元数据层也可以做到互相透视，数据可以做到自由流动。数据湖部分不只是支持EMR，也支持ESC Hadoop和云下IDC Hadoop。其中MaxCompute数据仓库也可以和数据湖EMR做一个数据打通，在用MaxCompute跟联播数据源做一个联播查询，这样可以把所有的数据在MaxCompute中做一个汇总。比如有三张表，在RDS和Hive中，同事MaxCompute里有大量事实表，如果需求是对这个三个表做一个联合查询，通过这个架构，可以很方便的做到这一点。更快的业务洞察DataWorks 自助开通湖仓一体：5分钟打通异构数据平台（Hadoop/ DLF+OSS ） 更广泛的生态对接支持对接阿里云云原生数据湖构建（DLF）支持查询 DeltaLake、Hudi 格式支持对接更多外部联邦数据源 Hologres （RDS、HBaseUpcoming！ ）更高的性能智能 Cache 实现 OSS/ HDFS 数据访问加速湖数据查询加速更好的综合数据开发与治理跨引擎开发和调度 统一湖/仓数据开发体验统一湖/仓全局资产管理湖仓一体的架构首先看右侧部分，是跟OSS和DLF侧的打通，因为在OSS 上我们归档大量的半结构化和结构化的数据。有关系型数据库，有nosql数据库，可以通过DLF组件把OSS上的元数据爬取出来。相比于在MaxCompute上创建OSS外表访问OSS数据，现在通过DLF统一自动识别OSS schema，MaxCompute直接映射这种方式，会更方便一些。一些数据的更新，schema的变更，DLF也可以自动识别到。同时DLF里面有用户权限管理，后续会陆续上线。也就是说对于OSS对接的引擎，统一的数据访问权限会收敛到DLF里。左侧是对Hadoop生态系统的打通，Hadoop包括阿里云半托管的EMR，开源的on ECS和IDC Hadoop，也包含主流的发行版CDH，也都可以做打通。下方再加上联邦数据源。通过MaxCompute可以连接到各种各样的云上数据源，通过上传DataWorks做统一的开发和管理，以及数据治理。 这样就有一个全域数据资产视图，开发工作数据也能联通，元数据也能投射到DataWorks之上。这就是整个湖仓一体的架构。二、相关产品介绍数据湖构建DataLakeForamtionDLF主要是针对OSS数据，做一个托管，上层对接其他引擎如EMR、Databricks、Datalnsight、PAI、MaxCompute、Hologres。这些引擎可以共享一份元数据在DLF上。后续企业可以根据不同部门，不同业务场景随意切换引擎。后期我们也会逐渐完善访问控制，以及一些企业级能力，比如安全，包括数据分层等。数据入湖支持多种数据源入湖，MySQL、SLS、OTS、Kafka等离线/实时入湖，支持Delta/Hudi等多种数据湖格式数据入湖预处理，支持字段mapping/转换/自定义udf操作元数据管理统一元数据管理，解决元数据多引擎一致性问题兼容开源生态API自动生成元数据，降低使用成本访问控制集中数据访问权限控制，多引擎统一集中式赋权数据访问日志审计，统计数据访问信息三、最佳实践Demo整体架构 企业构建和应用数据湖一般需要经历数据入湖、数据湖存储与管理、数据探索与分析等几个过程。主要介绍基于阿里云 MaxCompute、数据湖构建（DLF）+ 对象存储（OSS）构建一站式的数据入湖与分析实战。其主要数据链路如下：第一步：RDS数据准备RDS 数据准备，在 RDS 中创建数据库 academy_db。在账户中心创建能够读取 employees 数据库的用户账号。通过DMS登录数据库，运行一下语句创建 anti_fraud_result 表，及插入少量数据。CREATE TABLE `anti_fraud_result` ( `transactionid` varchar(32) NOT NULL, `uid` varchar(32) DEFAULT NULL, `card` varchar(32) DEFAULT NULL, `longitude` double(12,8) DEFAULT '12.00000000', `latitude` double(12,8) DEFAULT '12.00000000', PRIMARY KEY (`transactionid`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 ; INSERT INTO `anti_fraud_result` values ("12489571","82005","123123",3.14592040,101.12315432); INSERT INTO `anti_fraud_result` values ("12489572","82005","123123",3.14592040,101.12315432); INSERT INTO `anti_fraud_result` values ("12489573","82005","123123",3.14592040,101.12315432); INSERT INTO `anti_fraud_result` values ("12489574","82005","123123",3.14592040,101.12315432);第二部：DLF数据入湖进入DLF控制台界面：https://dlf.console.aliyun.com/cn-hangzhou/home，点击菜单 数据入湖 -> 数据源管理-> 新建数据源。输入数据库相关连接信息填写 RDS 实例、登录信息以及网络和安全组信息后，点击“连接测试”在 OSS 中新建 Bucket 和目录，如 bucket: rtcompute，目录：oss://rtcompute/csvfile/在 DLF 控制台界面点击左侧菜单“元数据管理”->“元数据库”，点击“新建元数据库”。填写名称 covid-19，新建目录并选择。创建入湖任务： a. DLF 控制台中点击菜单“数据入湖”->“入户任务管理”，点击“新建入湖任务” -> 选择“关系数据库实时入湖”，按照下图的信息填写数据源、目标数据湖、任务配置等信息。并保存。如果是 Hudi 格式，在“存储格式”中可以选择 Hudi： b. 下一步输入任务实例名称，RAM角色以及最大资源使用量，点击下一步确认信息 c. 在 DLF 控制台“入户任务管理” 界面中，找到刚创建的入湖任务列表，点击“运行”启动入湖作业；该数据入湖任务，属于全量+增量入湖，大约3至5分钟后，全量数据会完成导入，随后自动进入实时监听状态。如果有数据更新，则会自动更新至 Delta Lake 或 Hudi 数据格式中。第三步：MaxCompute中查询数据 开通 DataWorks 与 MaxCompute （快速启动请参考 文档），并进入 DataWorks 临时查询界面（请参考文档）创建 MaxCompute 外部 Project 映射 DLF 元数据：create externalproject -source dlf -name ext_dlf_delta -ref ms_proj1 -comment "DLF" -region "cn-shanghai" -db covid_19 -endpoint "dlf-share.cn-shanghai.aliyuncs.com" -ossEndpoint "oss-cn-shanghai-internal.aliyuncs.com"; -- 显示映射表信息 show tables in ext_dlf_delta;查询数据：set odps.sql.hive.compatible=true; set odps.sql.split.hive.bridge=true; select * from ext_dlf_delta.sh_rds;第四部：RDS中新增数据INSERT INTO `anti_fraud_result` values ("12489575","82005","123123",3.14592040,101.12315432); INSERT INTO `anti_fraud_result` values ("12489576","82005","123123",3.14592040,101.12315432); INSERT INTO `anti_fraud_result` values ("12489577","82005","123123",3.14592040,101.12315432); INSERT INTO `anti_fraud_result` values ("12489578","82005","123123",3.14592040,101.12315432); INSERT INTO `anti_fraud_result` values ("12489579","82005","123123",3.14592040,101.12315432);MaxCompute中校验数据set odps.sql.hive.compatible=true; set odps.sql.split.hive.bridge=true; select * from ext_dlf_delta.sh_rds;在1分钟之内，在 MaxCompute 中查看数据已更新：更多关于大数据计算、云数据仓库技术交流，欢迎扫码查看咨询。