使用Java将数据流式传输到HPCC

高性能计算集群（HPCC）是类似于Hadoop的分布式处理框架，除了它运行以自己的称为企业控制语言（ECL）的特定领域语言（DSL）编写的程序外。 ECL很棒，但是偶尔您会想用其他语言来执行繁重的任务。例如，您可能想利用Java编写的NLP库。

此外，HPCC通常针对类似于HDFS的文件系统上的数据进行操作。就像使用HDFS一样，一旦您超越了日志文件处理和静态数据快照的范围，您就会Swift产生对数据库后端的渴望。

实际上，我想说这是一个普遍的行业趋势：HDFS-> HBase，S3-> Redshift等。最终，您希望减少分析的延迟（接近零）。为此，您需要设置某种分布式数据库，该数据库能够支持批处理以及数据流/微分批处理。而且，您采用了一种不变/递增的数据存储方法，使您可以折叠基础结构，并在分析数据时将数据流传输到系统中（简化了处理过程）

但是我离题了，作为朝这个方向迈出的一步……

我们可以利用HPCC中的Java集成功能来支持Java中的用户定义函数。同样，我们可以利用相同的功能来添加其他后端存储机制（例如Cassandra）。更具体地说，让我们看一下HPCC / Java集成的流功能，以从外部源获取数据。

让我们首先看一下原始Java集成。

如果您具有HPCC环境设置，则Java集成将从/ opt / HPCCSystems / classes路径开始。您可以将类和jar文件拖放到该位置，并且可以从ECL中使用这些功能。请按照此页面上的说明进行操作。

如果遇到问题，请参阅该页面上的故障排除指南。最困难的部分是让HPCC查找您的班级。对我来说，我遇到了一个讨厌的JDK版本问题。默认情况下，HPCC在我的Ubuntu计算机上选择了旧的JDK版本。由于它使用的是旧版本，因此HPCC找不到使用“新” JDK（1.7）编译的类，这导致了一条模糊的消息：“无法解析类名”。如果遇到此问题，请拉出我提交的针对Ubuntu修复的补丁。

完成该工作后，您将可以使用以下语法从ECL调用Java：

IMPORT java;
integer add1(integer val) := IMPORT(java, 'JavaCat.add1:(I)I');
output(add1(10));

这非常好用，而且正如文档所建议的，如果数据复杂，则可以从Java方法返回XML。但是，如果您拥有大量的数据，而不是驻留在内存中，该怎么办？好吧，那么您需要将Java流传输到HPCC。 ;）

而不是从导入的方法返回实际数据，我们返回一个Java Iterator。然后，HPCC使用Iterator构造数据集。以下是一个示例Iterator。

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;public class DataStream implements Iterator {private int position = 0;private int size = 5;public static Iterator stream(String foo, String bar){return new DataStream();}@Overridepublic boolean hasNext() {position++;return (position < size);}@Overridepublic Row next() {return new Row("row");}@Overridepublic void remove() {}}

这是一个标准的Iterator，但请注意它返回一个Row对象，其定义如下：

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;public class Row {private String value;public Row(String value){this.value = value;}
}

该对象是一个Java bean。 HPCC将在映射到DATASET时设置成员变量的值。要确切了解这种情况如何发生，让我们看一下ECL代码：

IMPORT java;rowrec := recordstring value;
end;DATASET(rowrec) stream() := IMPORT(java, 'DataStream.stream:(Ljava/lang/String;Ljava/lang/String;)Ljava/util/Iterator;');output(stream());

在import语句之后，我们定义了一种称为rowrec的记录类型。在以下行中，我们导入UDF，然后将结果键入为包含rowrecs的DATASET。 rowrec中的字段名称必须与java bean上成员变量的名称匹配。 HPCC将使用迭代器，并使用next（）方法的返回值填充数据集。 ECL的最后一行输出返回的结果。

我已将以上所有代码提交给github存储库，其中包含一些有关使其运行的说明。玩得开心。

敬请期待更多…

想象一下，将这里概述的java流功能与将数据流出Cassandra的能力结合在一起，就像我之前的文章中所详细描述的那样。结果是一种强大的方法，可以使用Thor对存储在Cassandra中的数据（具有数据局部性！）运行批处理分析（可能对通过实时实时事件流获取的数据启用ECL作业！=）