写点什么

Apache Crunch:用于简化 MapReduce 编程的 Java 库

  • 2013-03-18
  • 本文字数:2454 字

    阅读完需:约 8 分钟

Apache Crunch(孵化器项目)是基于 Google 的 FlumeJava 库编写的 Java 库,用于创建 MapReduce 流水线。与其他用来创建 MapReduce 作业的高层工具(如 Apache Hive、Apache Pig 和 Cascading 等)类似,Crunch 提供了用于实现如连接数据、执行聚合和排序记录等常见任务的模式库。而与其他工具不同的是,Crunch 并不强制所有输入遵循同一数据类型。相反,Crunch 使用了一种定制的类型系统,非常灵活,能够直接处理复杂数据类型,如时间序列、HDF5 文件、Apache HBase 表和序列化对象(像 protocol buffer 或 Avro 记录)等。

Crunch 并不想阻止开发者以 MapReduce 方式思考,而是尝试使之简化。尽管 MapReduce 有诸多优点,但对很多问题而言,并非正确的抽象级别:大部分有意思的计算都是由多个 MapReduce 作业组成的,情况往往是这样——出于性能考虑,我们需要将逻辑上独立的操作(如数据过滤、数据投影和数据变换)组合为一个物理上的 MapReduce 作业。

本质上,Crunch 设计为 MapReduce 之上的一个薄层,希望在不牺牲 MapReduce 力量(或者说不影响开发者使用 MapReduce API)的前提下,更容易在正确的抽象级别解决手头问题。

尽管 Crunch 会让人想起历史悠久的 Cascading API,但是它们各自的数据模型有很大不同:按照常识简单总结一下,可以认为把问题看做数据流的人会偏爱 Crunch 和 Pig,而考虑 SQL 风格连接的人会偏爱 Cascading 和 Hive。

Crunch 的理念

PCollection 和 PTable<K, V> 是 Crunch 的核心抽象,前者代表一个分布式、不可变的对象集合,后者是 Pcollection 的一个子接口,其中包含了处理键值对的额外方法。这两个核心类支持如下四个基本操作:

  1. parallelDo:将用户定义函数应用于给定 PCollection,返回一个新的 PCollection 作为结果。
  2. groupByKey:将一个 PTable 中的元素按照键值排序并分组(等同于 MapReduce 作业中的 shuffle 阶段)
  3. combineValues:执行一个关联操作来聚合来自 groupByKey 操作的值。
  4. union:将两个或多个 Pcollection 看做一个虚拟的 PCollection。

Crunch 的所有高阶操作(joins、cogroups 和 set operations 等)都是通过这些基本原语实现的。Crunch 的作业计划器(job planner)接收流水线开发者定义的操作图,将操作分解为一系列相关的 MapReduce 作业,然后在 Hadoop 集群上执行。Crunch 也支持内存执行引擎,可用于本地数据上流水线的测试与调试。

有些问题可以从能够操作定制数据类型的大量用户定义函数受益,而 Crunch 就是为这种问题设计的。Crunch 中的用户定义函数设计为轻量级的,为满足应用程序的需要,仍然提供了完整的访问底层 MapReduce API 的功能。Crunch 开发者也可以使用 Crunch 原语来定义 API,为客户提供涉及一系列复杂 MapReduce 作业的高级 ETL、机器学习和科学计算功能。

Crunch 起步

可以从 Crunch 的网站下载最新版本的源代码或二进制文件,或者使用在 Maven Central 发布的 dependencies

源代码中有很多示例应用。下面是 Crunch 中 WordCount 应用的源代码:

复制代码
import org.apache.crunch.DoFn;
import org.apache.crunch.Emitter;
import org.apache.crunch.PCollection;
import org.apache.crunch.PTable;
import org.apache.crunch.Pair;
import org.apache.crunch.Pipeline;
import org.apache.crunch.impl.mr.MRPipeline;
import org.apache.crunch.type.writable.Writables;
public class WordCount {
public static void main(String[] args) throws Exception {
// Create an object to coordinate pipeline creation and execution.
Pipeline pipeline = new MRPipeline(WordCount.class);
// Reference a given text file as a collection of Strings.
PCollection<String> lines = pipeline.readTextFile(args[0]);
// Define a function that splits each line in a PCollection of Strings into a
// PCollection made up of the individual words in the file.
PCollection<String> words = lines.parallelDo(new DoFn<String, String>() {
public void process(String line, Emitter<String> emitter) {
for (String word : line.split("\\s+")) {
  emitter.emit(word);
}
}
}, Writables.strings()); // Indicates the serialization format
// The count method applies a series of Crunch primitives and returns
// a map of the top 20 unique words in the input PCollection to their counts.
// We then read the results of the MapReduce jobs that performed the
// computations into the client and write them to stdout.
for (Pair<String, Long> wordCount : words.count().top(20).materialize()) {
System.out.println(wordCount);
}
}
}

Crunch 优化方案

Crunch 优化器的目标是尽可能减少运行的 MapReduce 作业数。大多数 MapReduce 作业都是 IO 密集型的,因此访问数据的次数越少越好。公平地说,每种优化器(Hive、Pig、Cascading 和 Crunch)的工作方式本质上是相同的。但与其他框架不同的是,Crunch 把优化器原语暴露给了客户开发人员,对于像构造 ETL 流水线或构建并评估一组随机森林模型这样的任务而言,构造可复用的高阶操作更容易。

结论

Crunch 目前仍处于 Apache 的孵化器阶段,我们非常欢迎社区贡献(参见项目主页)让这个库更好。特别的是,我们正在寻求更高效的MapReduce 编译思想(包括基于成本考虑的优化)、新的MapReduce 设计模式,还希望支持更多的数据源和目标,如HCatalog、Solr 和ElasticSearch 等。还有很多把Crunch 带向如 Scala Clojure 等其他 JVM 语言的项目,也有很多使用 Crunch以R 语言来创建MapReduce 流水线的工具。

关于作者

Josh Wills 是 Cloudera 的数据科学主管,主要负责与客户和工程师一起基于 Hadoop 为不同行业开发解决方案。他从杜克大学获得数学专业学士学位,又从得克萨斯大学奥斯汀分校获得运筹学专业硕士学位。

查看英文原文: Apache Crunch: A Java Library for Easier MapReduce Programming

2013-03-18 17:068898
用户头像
臧秀涛 略懂技术的运营同学。

发布了 300 篇内容, 共 144.0 次阅读, 收获喜欢 35 次。

关注

评论

发布
暂无评论
发现更多内容

【YashanDB知识库】如何解决共享集群部署遇到报错:YAS-05721 invalid input parameter, reason: node name invalid.

YashanDB

数据库

京东零售数据可视化平台产品实践与思考

京东零售技术

数据可视化

“慢”增长时代的企业数据体系建设:超越数据中台

京东零售技术

大数据

【2024丨CSIG青年科学家会议 AI可信论坛】视觉内容安全技术的前沿进展与应用

热爱编程的小白白

多推理几步,生成式AI会变得更聪明吗?

JustYan

人工智能 大模型 生成式AI

京东科技基于 Apache SeaTunnel 探索复杂场景适配

白鲸开源

Clickhouse 数据同步 数据集成 Apache SeaTunnel 京东科技

2024年低代码趋势洞察——企业最看重的功能有哪些

JeeLowCode低代码平台

低代码开发 低代码开发工具 低代码,

Taro小程序开发性能优化实践

京东零售技术

taro

云交易技术对接全景

京东零售技术

云交易

Python for 和 while 循环:掌握循环控制的基本操作

敲代码不忘补水

Python 科技 while循环 for in 计算机科学与技术

Python 数据类型详解:列表、字典、元组与集合的操作指南

敲代码不忘补水

Python 科技 数据类型 tuple 计算机科学与技术

华为云Flexus X实例部署安装HivisionIDPhoto一个轻量级的AI证件照制作算法

平平无奇爱好科技

华为云Flexus X实例部署安装HivisionIDPhoto一个轻量级的AI证件照制作算法

平平无奇爱好科技

《计算机组成及汇编语言原理》阅读笔记:p82-p85

codists

【双喜】矩阵起源子公司荣膺“2024上海市高成长百家”第一、 荣获上海市技术创新资金立项资助

MatrixOrigin

AI 技术创新 MatrixOrigin 矩阵起源 GenAI

【YashanDB知识库】如何限制用户session连接数

YashanDB

数据库 yashandb

直击面试!阿里技术官手码12W字面试小册在Github上爆火

Summer

Java 程序员 面试 架构师 大厂

打造两轮差速机器人fishbot:从零开始构建移动机器人

芯动大师

ROS imu joint

矩阵起源荣膺"2024AI最佳实践案例"------多模态AI数据智能平台,让您的数据成就您的AI

MatrixOrigin

数据库 AI InfoQ 技术创新 MatrixOrigin

【YashanDB知识库】启动数据库时报错:YAS-02059 control file version 0.2.64 is incompatible with YashanDB version 0.2.65

YashanDB

数据库 yashandb

Hume 语音模型 OCTAVE:实现情感语音合成、声音克隆和多角色对话生成;通义开源多模态说话人识别项目 3D-Speaker

声网

【YashanDB知识库】如何处理报错"UDT column batch insert" has not been implemented yet

YashanDB

数据库 yashandb

【YashanDB知识库】如何排查YMP报错:”OCI版本为空或OCI的架构和本地系统的架构不符“

YashanDB

数据库 yashandb

【教程】第十一章 子任务 & 工时——化繁为简

NocoBase

开源 项目管理 低代码 教程 任务管理

最新2025整理Java面试八股文,大厂必备神器

Summer

Java 程序员 面试 架构师 大厂

Apache Crunch:用于简化MapReduce编程的Java库_语言 & 开发_Josh Wills_InfoQ精选文章