Clojure + Lemur

Question

我正在尝试使用lemur+clojure运行一些多步骤作业。

我在将多个输入作为参数传递给clojure+lemur时遇到了问题。

作为作业的第一步，我尝试运行emr流作业

lemur运行${CONF_DIR}/ run -pipeline.clj --主实例类型${MASTER_INSTANCE_TYPE} --从实例类型${SLAVE_INSTANCE_TYPE} --num实例${NUM_INSTANCES} --ami版本${AMI_VERSION} --hadoop版本${HADOOP_VERSION}--存储桶${BUCKET} --jar-src-path ${CONF_DIR}/run-pipeline.clj --输入文件夹"${input_folder}“--输出文件夹"${output_folder}”--缩减任务"${REDUCE_TASKS}“--映射任务"${MAP_TASKS}”

对于单个输入文件，我的代码如下所示

(import com.amazonaws.services.elasticmapreduce.util.StepFactory)
(import com.amazonaws.services.elasticmapreduce.model.StepConfig)
(import com.amazonaws.services.elasticmapreduce.util.StreamingStep)


(defn create-step-parsing [eopts]
 (def step (new StreamingStep))
 (.withInputs step (into-array [(str (:input-folder eopts) "/inputs/*")]))
 ...

这可以很好地工作，但是当我尝试传递输入文件列表时，我得到了错误

lemur运行${CONF_DIR}/ run -pipeline.clj --主实例类型${MASTER_INSTANCE_TYPE} --从实例类型${SLAVE_INSTANCE_TYPE} --num实例${NUM_INSTANCES} --ami版本${AMI_VERSION} --hadoop版本${HADOOP_VERSION}--存储桶${BUCKET} --jar-src-path ${CONF_DIR}/run-pipeline.clj --输入文件夹"${input_folder1}“--输入文件夹"${input_folder2}”--输入文件夹"${input_folder3}“--输入文件夹"${input_folder}”--输出文件夹"${output_folder}“--缩减任务"${REDUCE_TASKS}”--映射任务"${MAP_TASKS}“

(defn create-normalizer-step [eopts]
  (def step (new StreamingStep))
  (.withInputs step (to-array (:input-folder eopts)))

下面是我得到的错误

15:44:05 Exception in thread "main" java.lang.ClassCastException
15:44:05    at java.lang.Class.cast(Class.java:2990)
15:44:05    at clojure.lang.Reflector.boxArg(Reflector.java:429)
15:44:05    at clojure.lang.Reflector.boxArgs(Reflector.java:462)
15:44:05    at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:57)
15:44:05    at clojure.lang.Reflector.invokeInstanceMethod(Reflector.java:30)
15:44:05    at run_pipeline17$create_normalizer_step.invoke(run-pipeline.clj:18)
15:44:05    at run_pipeline17$run_pipeline.invoke(run-pipeline.clj:209)
15:44:05    at lemur.core$fire_BANG_.doInvoke(core.clj:711)
15:44:05    at clojure.lang.RestFn.invoke(RestFn.java:423)
15:44:05    at run_pipeline17$eval178.invoke(run-pipeline.clj:222)
15:44:05    at clojure.lang.Compiler.eval(Compiler.java:6465)
15:44:05    at clojure.lang.Compiler.load(Compiler.java:6902)
15:44:05    at clojure.lang.Compiler.loadFile(Compiler.java:6863)
15:44:05    at clojure.lang.RT$3.invoke(RT.java:305)
15:44:05    at lemur.core$execute_jobdef.invoke(core.clj:742)
15:44:05    at lemur.core$_main$fn__1388.invoke(core.clj:929)
15:44:05    at lemur.core$_main.doInvoke(core.clj:924)
15:44:05    at clojure.lang.RestFn.applyTo(RestFn.java:137)
15:44:05    at lemur.core.main(Unknown Source)

我添加的代码是从第17行到第19行。

谢谢

Answer 1

我认为问题出在您使用的数组函数上。在第一个工作示例中，您正在使用(into-array)。

从文档中：

(入数组aseq)

(into-array类型aseq)

返回组件设置为aseq中的值的数组。数组的组件类型是type (如果提供)，或者aseq中第一个值的类型(如果存在)，或者Object。aseq中的所有值必须与组件类型兼容。基元类型的类对象可以使用例如Integer/TYPE来获得。

因为您创建了字符串作为into-array的参数，所以clojure正确地为您创建了一个"string“数组。

在第二个不完整的示例中，使用(to-array)。从文档中：

(到数组列)

返回包含coll内容的对象数组，该数组可以是任何Collection。映射到java.util.Collection.toArray()。

在此实例中，您创建了一个“Object”数组。根据AWS Java API for StreamingStep，这与"String“的类型不兼容。

我的建议是使用(into-array)，类型规范为String。例如：(into-array String "hello“"goodby")。在我看来，显式的类型规范对于将来的阅读来说是很好的。但正如您在文档中看到的，(into-array)将为您猜测类型。