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

hpcc

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

此外&＃xff0c;HPCC通常对类似于HDFS的文件系统上的数据进行操作。 就像HDFS一样&＃xff0c;一旦超越了日志文件处理和静态数据快照的范围&＃xff0c;您就会很快对数据库后端产生需求。

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

但是我离题了&＃xff0c;作为朝这个方向迈出的一步……

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

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

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

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

完成该工作后&＃xff0c;您将可以使用以下语法从ECL调用Java&＃xff1a;

IMPORT java;
integer add1(integer val) :&＃61; IMPORT(java, &＃39;JavaCat.add1:(I)I&＃39;);
output(add1(10));

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

返回一个Java迭代器&＃xff0c;而不是从导入的方法返回实际数据。 然后&＃xff0c;HPCC使用Iterator构造数据集。 以下是一个示例Iterator。

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;public class DataStream implements Iterator {private int position &＃61; 0;private int size &＃61; 5;public static Iterator stream(String foo, String bar){return new DataStream();}&＃64;Overridepublic boolean hasNext() {position&＃43;&＃43;;return (position

这是一个标准的Iterator&＃xff0c;但请注意它返回一个Row对象&＃xff0c;其定义如下&＃xff1a;