Flume-Core

本章讲述Flume项目的核心表示层: 逻辑执行计划. 逻辑执行计划的执行请参见 算子执行框架

核心概念

Flume将分布式计算过程抽象为数据流在算子上的变换, 这就引申出了两个Flume中的核心概念: 算子和算子应用的范围. 接下来我们分别讲解这些概念.

Processor

在Flume中, 最核心的算子是Processor, 它接受N路数据流, 产出1路数据流. 我们这里举两个例子.

../_images/filter_processor.png

上图描述了FilterProcessor的逻辑, 它过滤掉的输入流X中的奇数数据, 得到一个新的数据流O.

../_images/cartesian_processor.png

上图描述了CartesianProcessor的逻辑, 它有两路输入X Y, 输出是X和Y中数据的笛卡尔积.

Processor可以被串联/级联组成Dag, 如下图所示:

../_images/processor_dag.png

通过上图我们可以看到, Processor之间可以任意串联, 一个Processor后面也可以跟随任意个后继Processor, 然后后继的每个Processor在该路上得到的输入都是一样的, 比如B和C得到的输入就完全一样.

Group/Scope

Processor算子是Flume中的基本计算单元, 它体现了分布式计算中的’计算’, 我们需要另外的概念来描述’分布式’. 考虑下面的例子

  1. select week, weekday, sum(sales_value)
  2. from sales_log
  3. group by rollup(week, weekday)
  4. orderby week, weekday

rollup是Oracle SQL数据库的概念, 含义是层级聚集, 具体到上面的例子, 就是分别以week和(week,weekday)作为分组应用进行求和, 其结果形如:

周天销售额
114363.55
124794.76
134718.25
145387.45
155027.34
1 24291.35
215652.84
224583.02
235555.77
245936.67
254508.74
2 26237.04
A 50528.39

我们看到, 所有这些输出都是SumProcessor的计算结果, 区别在于该算子应用的范围. 因此在Flume的定义中, 每个算子以Group为基本处理单位, 每次处理一组数据. 又因为分组是嵌套的, 如在{第一周销售记录}这一组数据, 又可按照工作日继续细分, 我们用Scope来面表达相同地位分组的集合. 每一个算子, 都必须置于某一个Scope下, 处理该Scope下所有Group的数据. 按照这种思路, 我们用如下图所示的逻辑执行计划来描述上述SQL的计算逻辑:

../_images/sales_sum_plan.png

在Flume中, 我们用Shuffle这个名词来表达分组和排序. 一个Shuffle过程可以有N路输入源参与, 其结果表示为一个嵌套的Scope和其中的N个ShuffleNode, 每个ShuffleNode有一路输入和一路输出, 输入是参与Shuffle过程的输入源, 其输出可以被同Scope的后续算子操作.

逻辑执行计划

在上一节中, 我们讲述了Flume计算抽象的基本原则. 本节我们要讲述完整的核心表示层：逻辑执行计划. 首先我们介绍逻辑执行计划的各个组成部分.

Node

  1. 组成逻辑计划的基本单位
  2. Node上绑定着相应的自定义算子
  3. 每个Node上只有一路逻辑上的输出
  4. 逻辑执行计划上的所有Node组成一个DAG

Scope/Group

  1. Scope是地位相同的各个数据分组的集合
  2. Scope中的数据按key分组, 每组为一个Group
  3. 逻辑计划上的每个Node都属于某个Scope
  4. 逻辑计划中的所有Scope组成一棵树

算子/Entity

  1. 算子是可以被反射的C++类, 用以承载用户逻辑. 具体参见 flume/core/entity.h
  2. 在所有算子中, 数据流被抽象为无类型的, 每条数据的类型都是void*
  3. 每个带有输出的Node上必须指定Objector算子, 该算子负责完成数据的序列化和反序列化工作

逻辑执行计划在 flume/proto/logical_plan.proto和entity.proto中定义, 下面给出以上三个概念的proto定义:

  1. message PbLogicalPlan {
  2.     repeated PbLogicalPlanNode node = 1;
  3.     repeated PbScope scope = 2;
  4. }
  6. message PbLogicalPlanNode {
  7.     enum Type {
  8.         UNION_NODE = 0;
  9.         LOAD_NODE = 1;
  10.         SINK_NODE = 2;
  11.         PROCESS_NODE = 3;
  12.         SHUFFLE_NODE = 4;
  13.     }
  15.     required string id = 1;  // uuid
  16.     required Type type = 2;
  17.     optional bytes debug_info = 3;
  18.     optional PbEntity objector = 4;
  19.     required string scope = 5;  // scope uuid
  20.     optional bool cache = 6 [default = false]; // indicate if the node need to be cached
  22.     // according to type, only corresponding member makes sense
  23.     optional PbUnionNode union_node = 100;
  24.     optional PbLoadNode load_node = 101;
  25.     optional PbSinkNode sink_node = 102;
  26.     optional PbProcessNode process_node = 103;
  27.     optional PbShuffleNode shuffle_node = 104;
  28. }
  30. message PbScope {
  31.     enum Type {
  32.       DEFAULT = 0;
  33.       INPUT = 1;  // only accept single LOAD_NODE under this scope
  34.       GROUP = 2;  // accept BROADCAST/KEY SHUFFLE_NODE under this scope
  35.       BUCKET = 3;  // accept BROADCAST/SEQUENCE SHUFFLE_NODE under this scope
  36.     };
  38.     required string id = 1;  // uuid
  39.     optional Type type = 2 [default = DEFAULT];  // declare optional for compatible reason
  40.     optional string father = 3;  // scope uuid
  41.     optional bool is_sorted = 4 [default = false];
  42.     optional uint32 concurrency = 5 [default = 1];  // a hint for execution
  44.    optional PbInputScope input_scope = 101;
  45.    optional PbGroupScope group_scope = 102;
  46.    optional PbBucketScope bucket_scope = 103;
  47. }
  49. message PbEntity { // PbEntity 是可被反射的C++类，可通过类名和config参数来得到相应实例
  50.   required string name = 1;
  51.   required bytes config = 2;
  52. }

下面分别介绍各个逻辑计划节点.

PROCESS_NODE

PROCESS_NODE是基本的数据处理节点, N入单出, 下面是它的proto定义:

  1. message PbProcessNode {
  2.     message Input {
  3.         required string from = 1;  // 上游来源节点
  5.         optional bool is_partial = 101 [default = false]; // 表示是否需要拥有全量数据才能计算（可用于map阶段的预聚合）
  6.         optional bool is_prepared = 102 [default = false]; // 表示输入是一个Stream还是一个Collection
  7.     };
  9.     repeated Input input = 1; // 输入的属性
  10.     required PbEntity processor = 2; // 实际的处理逻辑，可由用户自定义
  12.     optional int32 least_prepared_inputs = 101 [default = 0]; // 表示至少有几路输入是Collection才能开始计算
  13. }

Processor的接口定义如下:

  1. class Processor {
  2. public:
  3.     virtual ~Processor() {}
  4.     virtual void Setup(const std::string& config) = 0;
  6.     // keys传入该record所在Group在Scop中的位置
  7.     // inputs用来传入可迭代的输入.
  8.     // emitter用来将结果传给下游节点.
  9.     virtual void BeginGroup(const std::vector<toft::StringPiece>& keys,
  10.                             const std::vector<Iterator*>& inputs,
  11.                             Emitter* emitter) = 0;
  13.     // index表示传入的记录属于哪路输入.
  14.     // 对于第N路输入, 如果inputs[N] != NULL, 则index != N
  15.     virtual void Process(uint32_t index, void* object) = 0;
  17.     // 当前分组处理结束
  18.     virtual void EndGroup() = 0;
  19. };
  21. // Emitter are used to pass result to subsequent execution node.
  22. class Emitter {
  23. public:
  24.     virtual ~Emitter() {}
  26.     // Emit result to subsequent execution node. When Emit returns, object is not needed any more.
  27.     // If Emit returns false, means no more records is needed by subsequence nodes.
  28.     virtual bool Emit(void *object) = 0;
  30.     // No more outputs. User can explicit calling it to cancel execution.
  31.     virtual void Done() = 0;
  32. };
  34. class Iterator {
  35. public:
  36.     virtual ~Iterator() {}
  38.     virtual bool HasNext() const = 0;
  39.     virtual void* NextValue() = 0;
  40.     virtual void Reset() = 0;
  41.     virtual void Done() = 0;
  42. };

每个PROCESS_NODE都属于一个Scope, 该Scope存在多组数据. 从每个Processor对象的角度看, 在其生命期内, 其时序关系如下图所示:

../_images/process_cycle.png

每次Begin-Group周期该Processor处理其所在Scope的一个分组.

LOAD_NODE

LOAD_NODE代表了框架的输入, 每个Load操作会在Global Scope下创建一个新Scope和该Scope下的一个LOAD_NODE. 之所以Load操作会创建Scope, 是因为很多情况下数据是按组存储的, 比如UDW存储中的partition概念.

LOAD_NODE的proto定义如下:

  1. message PbLoadNode {
  2.     repeated string uri = 1;  // 确定数据源所在位置，可以有多个。
  3.     required PbEntity loader = 2; // 读取数据采取的方式，类似Hadoop的InputFormat
  4. }

Loader算子的定义比较接近与Hadoop中InputFormat的设定, 分为且分和读取两个部分. 不同的是, Loader算子并不默认输入都是KeyValue形式的. 另外, 在执行的时候, 每个split都是其数据所在分组的key. Loader算子的接口定义如下:

  1. class Loader {
  2. public:
  3.     virtual ~Loader() {}
  4.     virtual void Setup(const std::string& config) = 0;
  6.     // 参数uri指定了数据所在路径，splits是存放了数据切片后的结果,
  7.     // 如URI、数据起始位置和偏移量. splits中的每个元素， 都会作为Load的参数
  8.     virtual void Split(const std::string& uri, std::vector<std::string>* splits) = 0;
  10.     // split是对一个数据块的描述，是 Split() 方法存入的.
  11.     // 这个方法中是对这个数据块的具体处理逻辑，比如反序列化和简单过滤等。
  12.     virtual void Load(const std::string& split, Emitter* emitter) = 0;
  13. };

SINK_NODE

SINK_NODE代表了框架的输出. 和PROCESS_NODE相同, SINK_NODE也属于某个Scope, 将该组Scope中的每组数据输出到外部系统. SINK_NODE的proto定义如下:

  1. message PbSinkNode {
  2.     required string from = 1; // 指数据所在位置，可以实现每个scope有不同的输出
  3.     required PbEntity sinker = 2; // 指输出数据是采取的方式，类似Hadoop的OutputFormat
  4. }

Sinker的接口定义如下:

  1. class Sinker {
  2. public:
  3.     virtual ~Sinker() {}
  4.     virtual void Setup(const std::string& config) = 0;
  6.     // 打开要写入的文件, 相当于Processor中的BeginGroup.
  7.     virtual void Open(const std::vector<toft::StringPiece>& keys) = 0;
  9.     // 写入实际数据。每条记录都是void*类型，由用户自己转换.
  10.     virtual void Sink(void* object) = 0;
  12.     // 关闭写入. 具有Commit语义
  13.     virtual void Close() = 0;
  14. };

SHUFFLE_NODE

SHUFFLE_NODE代表分组后的数据流, 由Shuffle操作产生. 其proto定义如下所述:

  1. message PbShuffleNode {
  2.     // 数据源参与分组的三种方式.
  3.     enum Type {
  4.         BROADCAST = 0; // 不参与下面两种处理，所有的记录都会被分发到每一组中
  5.         KEY = 1;       // 表示按key分组，不同的key属于不同的组
  6.         SEQUENCE = 2;  // 表示将数据分桶，预先设定桶数，按照某种策略（如hash）将key分到这些桶中
  7.     };
  9.     required string from = 1; // 上游来源节点
  10.     required Type type = 2;
  11.     optional PbEntity key_reader = 3;  // 表示用来提取key的方式，对应于KEY类型的shuffle
  12.     optional PbEntity partitioner = 4;  // 表示分桶方式，对应于SEQUENCE类型的shuffle
  13. }

Flume支持两种分组方式, 按Key聚集和分桶. 按Key聚集是为参与分组的每条记录附加一个key, 把所有key的记录汇聚到同一组中. 分桶是指事先决定好分组数量, 再把每条记录分配到某个桶中的分组方式.

这个过程中涉及到KeyReader和Partitioner两种算子, 其接口定义如下:

  1. class KeyReader {
  2. public:
  3.     virtual ~KeyReader() {}
  5.     virtual void Setup(const std::string& config) = 0;
  7.     // 具体的提取key的逻辑实现，object是整条记录，由用户自己理解其类型.
  8.     // buffer是最终存放key的变量，要求必须将key转换为char* 存放到buffer中。
  9.     // wing/common 下的comparable.h中提供了专门方法，生成可用来排序的string类型的key。
  10.     // 同时提供了升序和降序两种方法。
  11.     // 返回值是key的实际长度。
  12.     virtual uint32_t ReadKey(void* object, char* buffer, uint32_t buffer_size) = 0;
  13. };
  15. class Partitioner {
  16. public:
  17.     virtual ~Partitioner() {}
  19.     virtual void Setup(const std::string& config) = 0;
  21.     // 返回该条记录应该属于的分桶.
  22.     virtual uint32_t Partition(void* object, uint32_t partition_number) = 0;
  23. };

UNION_NODE

UNION_NODE用来将多个数据源和合并为一个数据源统一处理, 其proto定义如下:

  1. message PbUnionNode {
  2.     repeated string from = 1;  // 用于合并数据流，repeated字段中存放的是多个上游节点。
  3. }

编程实例 - WordCount

这里我们实现一个小程序，用来计算在两个文件中出现频率最高的前50个词.

  1. namespace baidu {
  2. namespace flume {
  3. namespace runtime {
  4. namespace dce {
  6. using core::Emitter;
  7. using core::Iterator;
  8. using core::KeyReader;
  9. using core::LogicalPlan;
  10. using core::Objector;
  11. using core::Processor;
  13. // 自定义Loader, 从源文件中读取数据传给下游节点
  14. class TextLoader : public core::Loader {
  15. public:
  16.     virtual ~TextLoader() {}
  18.     virtual void Setup(const std::string& config) {}
  20.     virtual void Split(const std::string& uri, std::vector<std::string>* splits) {
  21.         splits->push_back(uri);
  22.     }
  24.     virtual void Load(const std::string& split, Emitter* emitter) {
  25.         std::ifstream fin(split.c_str());
  26.         std::string line;
  27.         Record record;
  28.         while (getline(fin, line)) {
  29.             record.key = "";
  30.             record.value = line;
  31.             emitter->Emit(&record);
  32.         }
  33.     }
  34. };
  36. // 用来切词，并对每个词的出现次数置为1
  37. class WordSplitter : public Processor {
  38. public:
  39.     WordSplitter() : m_emitter(NULL) {}
  41.     virtual ~WordSplitter() {}
  43.     virtual void Setup(const std::string& config) {}
  45.     virtual void BeginGroup(const std::vector<toft::StringPiece>& keys,
  46.                             const std::vector<Iterator*>& inputs,
  47.                             Emitter* emitter) {
  48.         m_emitter = emitter;
  49.     }
  51.     virtual void Process(uint32_t index, void* object) {
  52.         Record* record = static_cast<Record*>(object);
  53.         std::vector<std::string> words;
  54.         toft::SplitStringByAnyOf(record->value, " ,.?!:;~@#$%-&*+-()[]{}|<>/\\'\"\n\t\r", &words);
  55.         for (size_t i = 0; i < words.size(); ++i) {
  56.             WordSum::ValueType single;
  57.             single.word = words[i];
  58.             single.sum = 1;
  59.             m_emitter->Emit(&single);
  60.         }
  61.     }
  63.     virtual void EndGroup() {}
  65. private:
  66.     Emitter* m_emitter;
  67. };
  70. // 用来序列化反序列化的方法
  71. class WordSum : public Objector {
  72. public:
  73.     struct ValueType {
  74.         ValueType() : sum(0) {}
  75.         std::string word;
  76.         int sum;
  77.     };
  79.     virtual ~WordSum() {}
  81.     virtual void Setup(const std::string& config) {}
  83.     virtual uint32_t Serialize(void* object, char* buffer, uint32_t buffer_size) {
  84.         ValueType* value = static_cast<ValueType*>(object);
  85.         Record record;
  86.         record.key = value->word;
  87.         std::string sum_text = boost::lexical_cast<std::string>(value->sum);
  88.         record.value = sum_text;
  89.         return m_objector.Serialize(&record, buffer, buffer_size);
  90.     }
  92.     virtual void* Deserialize(const char* buffer, uint32_t buffer_size) {
  93.         Record* record = static_cast<Record*>(m_objector.Deserialize(buffer, buffer_size));
  94.         ValueType* value = new ValueType();
  96.         value->word = record->key.as_string();
  97.         value->sum = boost::lexical_cast<int>(record->value.as_string());
  98.         m_objector.Release(record);
  99.         return value;
  100.     }
  102.     virtual void Release(void* object) {
  103.         ValueType* value = static_cast<ValueType*>(object);
  104.         delete value;
  105.     }
  107. private:
  108.     RecordObjector m_objector;
  109. };
  111. // shuffle按key分组，提取分组的key，为单词
  112. class WordIdentity : public KeyReader {
  113. public:
  114.     virtual ~WordIdentity() {}
  116.     virtual void Setup(const std::string& config) {}
  118.     virtual uint32_t ReadKey(void* object, char* buffer, uint32_t buffer_size) {
  119.         WordSum::ValueType* single = static_cast<WordSum::ValueType*>(object);
  120.         if (single->word.size() <= buffer_size) {
  121.             memcpy(buffer, single->word.data(), single->word.size());
  122.         }
  123.         return single->word.size();
  124.     }
  125. };
  127. // 统计每个单词个数
  128. class WordCount : public Processor {
  129. public:
  130.     WordCount() : m_emitter(NULL) {}
  132.     virtual ~WordCount() {}
  134.     virtual void Setup(const std::string& config) {}
  136.     virtual void BeginGroup(const std::vector<toft::StringPiece>& keys,
  137.                             const std::vector<Iterator*>& inputs,
  138.                             Emitter* emitter) {
  139.         m_adder.word.clear();
  140.         m_adder.sum = 0;
  141.         m_emitter = emitter;
  142.     }
  144.     virtual void Process(uint32_t index, void* object) {
  145.         WordSum::ValueType* value = static_cast<WordSum::ValueType*>(object);
  146.         CHECK_EQ(1, value->sum);
  147.         if (m_adder.word.empty()) {
  148.             m_adder.word = value->word;
  149.             m_adder.sum = value->sum;
  150.         } else {
  151.             CHECK_EQ(m_adder.word, value->word);
  152.             m_adder.sum += value->sum;
  153.         }
  154.     }
  156.     virtual void EndGroup() {
  157.         Record record;
  158.         record.key = m_adder.word;
  159.         std::string sum_text = boost::lexical_cast<std::string>(m_adder.sum);
  160.         record.value = sum_text;
  161.         m_emitter->Emit(&record);
  162.     }
  164. private:
  165.     WordSum::ValueType m_adder;
  166.     Emitter* m_emitter;
  167. };
  169. // 按key分组，提取分组的key，为单词个数
  170. class WordNum : public KeyReader {
  171. public:
  172.     virtual ~WordNum() {}
  174.     virtual void Setup(const std::string& config) {
  175.     }
  177.     virtual uint32_t ReadKey(void* object, char* buffer, uint32_t buffer_size) {
  178.         Record* record = static_cast<Record*>(object);
  179.         std::string num;
  180.         int value = boost::lexical_cast<int>(record->value.as_string());
  181.         ::baidu::wing::AppendReverseOrdered(value, &num);
  182.         if (num.size() <= buffer_size) {
  183.             memcpy(buffer, num.data(), num.size());
  184.         }
  185.         return num.size();
  186.     }
  187. };
  189. // CORE部分还没提供 limit 算子，需要自己实现
  190. class Limit : public Processor {
  191. public:
  192.     Limit() : m_emitter(NULL) {}
  194.     virtual ~Limit() {}
  196.     virtual void Setup(const std::string& config) {
  197.         m_limit = boost::lexical_cast<int>(config);
  198.     }
  200.     virtual void BeginGroup(const std::vector<toft::StringPiece>& keys,
  201.                             const std::vector<Iterator*>& inputs,
  202.                             Emitter* emitter) {
  203.         m_emitter = emitter;
  204.     }
  206.     virtual void Process(uint32_t index, void* object) {
  207.         if (results.size() < m_limit) {
  208.             results.push_back(*(static_cast<Record*>(object)));
  209.         }
  210.     }
  212.     virtual void EndGroup() {
  213.         for (int i = 0; i < results.size(); i++) {
  214.             m_emitter->Emit(&results[i]);
  215.         }
  216.     }
  218. private:
  219.     std::vector<Record> results;
  220.     Emitter* m_emitter;
  221.     int m_limit;
  222. };
  224. // 写入文件使用的 Sinker
  225. class TextSinker : public core::Sinker {
  226. public:
  227.     virtual ~TextSinker() {}
  229.     virtual void Setup(const std::string& config) {
  230.         m_path = config;
  231.     }
  233.     virtual void Open(const std::vector<toft::StringPiece>& keys) {
  234.         m_fout.open(m_path.c_str());
  235.     }
  237.     virtual void Sink(void* object) {
  238.         Record* record = static_cast<Record*>(object);
  239.         m_fout << record->key << " " << record->value << endl;
  240.     }
  242.     virtual void Close() {
  243.         m_fout.close();
  244.     }
  245. private:
  246.     std::ofstream m_fout;
  247.     std::string m_path;
  248. };
  250. TEST(DceBackendTest, Test) {
  251.     LOG(WARNING) << "test begin!";
  252.     toft::scoped_ptr<LogicalPlan> plan(new LogicalPlan());
  254.     // Map部分
  255.     LogicalPlan::Node* single_word = plan
  256.         // 读入两个文件
  257.         ->Load("/home/work/xuekang/tmp/hamlet.txt", "/home/work/xuekang/tmp/hamlet-copy.txt")
  258.         // 指定Loader和Objector
  259.         ->By<TextLoader>()->As<RecordObjector>()
  260.         // 指定Processor和Objector
  261.         ->ProcessBy<WordSplitter>()->As<WordSum>();
  262.     // Shuffle部分，指定分发使用的 KeyReader
  263.     plan->Shuffle(plan->global_scope(), single_word)->node(0)->MatchBy<WordIdentity>()
  264.         // 指定Processor和Objector
  265.         ->ProcessBy<WordCount>()->As<RecordObjector>()
  266.         // 回到上层scope
  267.         ->LeaveScope()
  268.         // 对子scope的全量数据做排序，指定 KeyReader
  269.         ->SortBy<WordNum>()
  270.         // 指定自定义的 Limit Processor和 Objector
  271.         ->ProcessBy<Limit>("50")->As<RecordObjector>()
  272.         // 指定Sinker和写入的文件
  273.         ->SinkBy<TextSinker>("/home/work/xuekang/tmp/hamlet-output");
  275.     plan->RunLocally();
  276. }
  278. }  // namespace dce
  279. }  // namespace runtime
  280. }  // namespace flume
  281. }  // namespace baidu
  283. int main(int argc, char* argv[]) {
  284.     ::testing::InitGoogleTest(&argc, argv);
  285.     ::google::ParseCommandLineFlags(&argc, &argv, true);
  286.     ::baidu::flume::InitBaiduFlume();
  287.     return RUN_ALL_TESTS();
  288. }

Reference

本节讲述 LogicalPlan 类提供的各个接口.

Load

用于读入数据源，需要指定数据所在的路径。

有两种调用方法：传入包含所有路径的 vector， 或直接列出每个路径，最多支持4个，以下大多API都支持这两种方式。

组成逻辑计划时会在 Global Scope 中创建一个 Scope 和一个 LoadNode ，同时使用 By<Loader>() 方法指定具体使用的 Loader （可自定义）， As<Objector>() 方法指定 Objector （可自定义），表示本节点处理完成后传输给下游节点的数据格式。

Sink

用于将结果写入目标路径，调用时要指定写入时使用的Sinker（可自定义）和目的路径。

Process

具体的数据处理逻辑，用户可自己实现此方法。需要指定所属Scope，以及数据来源的节点。

Union

合并多条数据流, 用户只能控制使用的Objector, 合并操作由框架保证。

Shuffle

Shuffle节点, 新建一个scope, 存放为每一个来源节点创建的shuffle node。

shuffle node 可以通过 MatchAny() MatchBy() DistributeAll() 等方法设置不同的shuffle方式，需要指定KeyReader。

ToProtoMessage

将一个节点的相关信息写入pb的message，方便作为执行计划传递给runtime。

RunLocally

本地执行逻辑计划。

Run

远程模式执行逻辑计划，同时指定执行后端和运行资源。