1

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

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

作为我工作的第一步，我尝试运行 emr Streaming Job

狐猴运行 ${CONF_DIR}/run-pipeline.clj --master-instance-type ${MASTER_INSTANCE_TYPE} --slave-instance-type ${SLAVE_INSTANCE_TYPE} --num-instances ${NUM_INSTANCES} --ami-version ${ AMI_VERSION} --hadoop-version ${HADOOP_VERSION}--bucket ${BUCKET} --jar-src-path ${CONF_DIR}/run-pipeline.clj --input_folder "${input_folder}" --output-folder " ${output_folder}" --reduce-tasks "${REDUCE_TASKS}" --map-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/*")]))
 ...

这工作正常，但是当我尝试传递输入文件列表时出现错误

狐猴运行 ${CONF_DIR}/run-pipeline.clj --master-instance-type ${MASTER_INSTANCE_TYPE} --slave-instance-type ${SLAVE_INSTANCE_TYPE} --num-instances ${NUM_INSTANCES} --ami-version ${ AMI_VERSION} --hadoop-version ${HADOOP_VERSION}--bucket ${BUCKET} --jar-src-path ${CONF_DIR}/run-pipeline.clj --input_folder "${input_folder1}" --input_folder "${ input_folder2}" --input_folder "${input_folder3}" --input_folder "${input_folder}" --output-folder "${output_folder}" --reduce-tasks "${REDUCE_TASKS}" --map-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 行。

谢谢

4

1 回答 1

0

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

从文档：

（入阵列 aseq）

（入数组类型 aseq）

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

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

在您的第二个损坏的示例中，您使用 (to-array)。从文档：

（到数组 coll）

返回一个包含 coll 内容的 Objects 数组，它可以是任何 Collection。映射到 java.util.Collection.toArray()。

在本例中，您创建了一个“对象”数组。根据适用于StreamingStep的 AWS Java API，这与“字符串”类型不兼容。

我的建议是使用 (into-array)，类型规范为 String。例如：(into-array String ["hello" "goodby"])。在我看来，显式类型规范很好，以便将来阅读澄清。但是正如您在文档中看到的那样， (into-array) 会为您猜测输入。

于 2015-02-10T13:34:30.540 回答