诞生于2015年6月的Apache Apex可谓横空出世，其同样源自DataTorrent及其令人印象深刻的RTS平台，其中包含一套核心处理引擎，仪表板、诊断与监控工具套件外加专门面向数据科学家用户的图形流编程系统dtAssemble。
作为新的开源数据流剖析计划，Apex脱胎于DataTorrent的RTS渠道，可以带来超卓的速度体现并简化编程要求。
说起数据流剖析使命，咱们首要想到的自然是Spark。尽管Spark现已凭借着2.0版别将非结构化与结构化两类数据的剖析才能融为一体，但Storm的1.0版别处理了本身难于运用的问题。
诞生于2015年6月的Apache Apex可谓横空出世，其相同源自DataTorrent及其令人形象深入的RTS渠道，其间包括一套中心处理引擎，仪表板、确诊与监控东西套件外加专门面向数据科学家用户的图形流编程体系dtAssemble。
作为RTS渠道的中心处理引擎，Apex可以说是DataTorrent献给Apache的又一份大礼。Apex的规划方针在于运转咱们的现有Hadoop生态体系，并运用YARN完结按需规划弹性且通过HDFS完结容错才能。尽管其并不像RTS渠道那样功用全面，但Apex现已足以供给咱们期望数据处理渠道所能完结的大都主要功用。
Apex运用示例
下面咱们来看一套根本Apex流程示例，其间将触及多项中心概念。在本示例中，咱们将读取Kafka中的日志条目，对日志记载类型进行计数并将其写入控制台傍边。相关代码片段将实践列出，咱们也可以点击此处获取GitHub上的完好运用。
Apex的中心概念在于operator，其归于Java类，担任完结输入信息接纳与输出信息生成。(假如咱们了解Storm，那么其作用根本类似于bolt与spout。)别的，每个operator还会界说一组用于数据输入或输出的端口。该办法的实践作用在于读取来自InputPort的输入信息，或许通过OutportPort向下流发送数据。
通过operator的数据流将进行建模，行将数据流拆分为根据时刻的数据窗口——但与Spark的microbathcing不同，Apex中的输入数据处理无需等候窗口完毕即可开端进行。
DataTorrent
在以下示例中，咱们需求3个operator，它们各自对应三种Apex所支撑的operator类型中的一种：输入operator担任由Kafka读取信息条目，通用operator担任对日志类型进行计数，而输出opeartor则将其写入控制台。关于***种与第三种，咱们可以直接运用Apex的Malhar库，但在第二种中咱们需求运用定制化事务逻辑以对检查到的不同日志类型进行计数。
下面来看咱们的LogCounterOperator代码内容：
public class LogCounterOperator extends BaseOperator {
private HashMap counter;
public transient DefaultInputPort input = new DefaultInputPort() {
@Override
public void process(String text) {
String type = text.substring(0, text.indexOf(' '));
Integer currentCounter = counter.getOrDefault(type, 0);
counter.put(type, currentCounter+1);
}
};
public transient DefaultOutputPort> output = new DefaultOutputPort<>();
@Override
public void endWindow() {
output.emit(counter);
}
@Override
public void setup(OperatorContext context){
counter = new HashMap();
}
}
这儿咱们运用简略的HashMap进行日志类型计数，一起界说2个端口以通过该operator完结数据流处理：其一担任输入，其二担任输出。在输入过程中，不兼容operator将引发编译时失利。需求留意的是，尽管我在这儿只界说了1个输入端口与1个输出端口，但咱们也可根据需求界说多个端口。
通用opeartor的生命周期十分简略。Apex会首要调用 setup()以进行任何必要的初始化操作;在以上示例中， setup()担任完结HashMap的创立作业。其随后调用beginWindow()以声明新的输入处理窗口/批量使命正在开端，接着在整个过程中对各数据条目调用。假如当时窗口的剩余时刻归零，Apex则会调用endWindow()。咱们不需求任何针对单一窗口的逻辑，因而将BaseOperator中的beginWindow()界说留空即可。但是，在每个窗口的结尾，咱们都需求发送当时计数效果，然后将HashMap通过输出端口进行发送。
与此一起，通过重写的process()办法担任处理咱们的事务逻辑，即从日志行中提取***个词并更新计数器。***，咱们调用teardown()办法，然后确保Apex流程得到必要的整理——本示例其实并不需求整理，但出于演示的考虑，咱们将整理HashMap。
现在咱们的operator现已创立完结，接下来需求构建流程本身。假如咱们了解Storm拓扑结构，那么应该可以轻松了解以下代码：
public void populateDAG(DAG dag, Configuration conf) {
KafkaSinglePortStringInputOperator kafkaInput = dag.addOperator("KafkaInput", new KafkaSinglePortStringInputOperator());
kafkaInput.setIdempotentStorageManager(new IdempotentStorageManager.FSIdempotentStorageManager());
LogCounterOperator logCounter = dag.addOperator("LogCounterOperator", new LogCounterOperator());
ConsoleOutputOperator console = dag.addOperator("Console", new ConsoleOutputOperator());
dag.addStream("LogLines", kafkaInput.outputPort, logCounter.input);
dag.addStream("Console", logCounter.output, console.input);
}
咱们首要界说DAG(即operator)节点。之后，咱们界说图形鸿沟(在Apex词汇中称其为‘stream’)。这些stream担任将某一operator的输出端口接入另一opeartor的输入端口。在这儿，咱们将Kafka接入LogCounterOperator，并将输出端口接入ConsoleOutputOperator。作业完结!假如咱们编译并运转该运用，则可以在规范输出效果中看到HashMap：
{INFO=1}
{ERROR=1, INFO=1}
{ERROR=1, INFO=2}
{ERROR=1, INFO=2, DEBUG=1}
…
Malhar: 丰厚的有用组件
Operator的***优势在于其体积细巧且通过清晰界说，因而可以轻松完结构建与测验。其接合方法类似于乐高积木——专一的差异在于乐高积木是现成的，但operator需求咱们自行创立。
Malhar就像是一个巨大的乐高积木桶，其间旋转有很多规范的2 x 4根本件供咱们运用。无论是读取Splunk，在FTP站点上兼并文本文件信息仍是将效果存储在HBase傍边，Malhar都能协助咱们完结。
有了Malhar供给的丰厚operator组件，Apex就变得***吸引力，这意味着咱们只需求规划事务逻辑即可。有时候Malhar operator的阐明文档比较粗糙，但该库中的一切都配备有测验机制，因而咱们可以轻松检查不同组件间的协作作用。
Apex还供给其它一些超卓的规划效果。除了常见的目标与陈述计划外，dtCli运用答应咱们以动态方法改变运转时中的已提交运用。咱们是否期望向HDFS傍边增加一些担任写入日志条目的operator，但又不期望影响到运用的全体运转?Apex可以轻松完结这项使命。
开源数据流处理引擎现已适当丰厚，但要在其间锋芒毕露则绝非易事。跟着Malhar库供给的巨大opeartor选项以及Apex本身所具有的超卓容错才能、低推迟以及可扩展性，Apex现已成为一款速度超卓且可用于出产环境的抱负结构。
在这儿，我主张DataTorrent为Apache Beam开发一套Apex运转器，然后协助开发者们更轻松地将自己的运用从现有结构中移植出来。当然，Apex现在现已适当优异，足以成为值得咱们仔细考量的数据流处理引擎。
原文链接：Look out, Spark and Storm, here comes Apache Apex
