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

比眉伴天荒 2023-02-25 03:44 24阅读 0赞

hpcc

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

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

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

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

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

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

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

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

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

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

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

返回一个Java迭代器，而不是从导入的方法返回实际数据。 然后，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();
        }
    
        @Override
        public boolean hasNext() {
            position++;
            return (position < size);
        }
    
        @Override
        public Row next() {
            return new Row("row");
        }
    
        @Override
        public 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 := record
      string 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存储库][github] ，其中包含一些有关使其运行的说明。 玩得开心。

敬请期待更多…

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

> 翻译自: [https://www.javacodegeeks.com/2015/05/streaming-data-into-hpcc-using-java.html][https_www.javacodegeeks.com_2015_05_streaming-data-into-hpcc-using-java.html]

hpcc

[Link 1]: https://wiki.hpccsystems.com/display/hpcc/Java+Integration
[Ubuntu]: https://github.com/hpcc-systems/HPCC-Platform/pull/7284
[github]: https://github.com/boneill42/hpcc-java-streaming-example
[Link 2]: https://brianoneill.blogspot.com/2015/05/data-locality-w-cassandra-how-to-scan.html
[https_www.javacodegeeks.com_2015_05_streaming-data-into-hpcc-using-java.html]: https://www.javacodegeeks.com/2015/05/streaming-data-into-hpcc-using-java.html