java - Morphlines Java自定义命令类在哪里存储？

Question

我有 csv 文件，它们通过 Morphline Sink 在 Solr 中建立索引。我想在 Morphlines 中编写一个自定义 Java 函数，在将其中一个 csv 字段索引到 SOLR 之前对其进行哈希处理。我尝试改编这个例子:http://blog.cloudera.com/blog/2014/04/how-to-process-data-using-morphlines-in-kite-sdk/ (请参阅将数据导入 Hadoop 集群部分)但我不确定何时应该存储 Java 类。

因此，在本例中，公开可用的“百万首歌曲”数据集是从 Last.fm 中选取的。这个想法是选择给定日期之前和之后的歌曲。

新的Morphline命令的配置是:

morphlines : [
    {
        id : morphline1
        importCommands : ["org.kitesdk.**", "com.sequenceiq.lastfm.etl.**"]
        commands : [
            {
                readJson {
                    outputClass : java.util.Map
                }
            }
            {
                latestSongs {
                    field : timestamp
                    operator: >
                    pattern: "2011-08-03"
                }
            }
        ]
    }]

以及对应的Java代码:

@Override
protected boolean doProcess(Record record) {
    Map attachmentBody = (Map) record.get(Fields.ATTACHMENT_BODY).get(0);
    String fieldValue = attachmentBody.get(fieldName).toString();

    try {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        Date fieldDate = sdf.parse(fieldValue);
        Date patternDate = sdf.parse(pattern + " 00:00:00");

        if (operator.equals(HIGHER)) {
            if (!fieldDate.after(patternDate)) {
                return true;
            }
        } else if (operator.equals(LOWER)) {
            if (!fieldDate.before(patternDate)) {
                return true;
            }
        } else if (operator.equals(EQUALS)) {
            if (fieldDate.getYear() != patternDate.getYear()
                    || fieldDate.getMonth() != patternDate.getMonth()
                    || fieldDate.getDay() != patternDate.getDay()) {
                return true;
            }
        } else {
            LOG.info("bad operator syntax");
        }
    } catch (Exception e) {
        LOG.info("parse exception: " + e.getMessage());
        return false;
    }
    record.removeAll(Fields.ATTACHMENT_BODY);
    try {
        record.put(Fields.MESSAGE, OBJECTMAPPER.writeValueAsString(attachmentBody));
    } catch (JsonProcessingException e) {
        LOG.info("parse exception: " + e.getMessage());
        return false;
    }
    return super.doProcess(record);
}  

感谢任何帮助!

Answer 1

您需要将自定义java类编译成jar文件。例如通过“mvn package”，如下所示:https://github.com/kite-sdk/kite-examples/tree/master/kite-examples-morphlines

然后将 jar 文件添加到 java 类路径，例如通过将 jar 文件复制到 lib/ 目录。