hadoop - Clojure + 狐猴

Question

我正在尝试使用 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-版本 ${AMI_VERSION} --hadoop-version ${HADOOP_VERSION}--bucket ${BUCKET} --jar-src-path ${CONF_DIR}/run-pipeline.clj --input_folder "${input_folder}"--output -文件夹 "${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-版本 ${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-任务“${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 type aseq)

Returns an array with components set to the values in aseq. The array's component type is type if provided, or the type of the first value in aseq if present, or Object. All values in aseq must be compatible with the component type. Class objects for the primitive types can be obtained using, e.g., Integer/TYPE.

因为您要创建字符串作为 into-array 的参数，clojure 会正确地为您创建一个“string”数组。

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

(to-array coll)

Returns an array of Objects containing the contents of coll, which can be any Collection. Maps to java.util.Collection.toArray().

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

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