从 Composer 在 dataproc 集群上执行 bash 脚本

我想在使用简单的 shell 脚本创建集群后将 jar 添加到特定位置的 dataproc 集群中。

创建 dataproc 集群后，我想自动执行此步骤以从 Composer 运行，下一步是执行 bash 脚本，该脚本会将 jar 添加到 data proc 集群。

您能否建议使用哪个气流运算符在 dataproc 集群上执行 bash 脚本？

要在主节点上运行简单的 shell 脚本，最简单的方法是使用pig sh https://pig.apache.org/docs/latest/cmds.html#shDataproc 作业，如下所示：

gcloud dataproc jobs submit pig --cluster ${CLUSTER} --execute 'sh echo hello world'

或使用pig fs https://pig.apache.org/docs/latest/cmds.html#fs直接复制jar文件：

gcloud dataproc jobs submit pig --cluster ${CLUSTER} --execute 'fs -cp gs://foo/my_jarfile.jar file:///tmp/localjar.jar'

这些 gcloud 命令的等效 Airflow 操作员设置将使用DataProcPig操作符 https://github.com/apache/airflow/blob/master/airflow/contrib/operators/dataproc_operator.py#L670与query字符串参数

如果您需要将 jar 文件放置在所有节点上，最好只使用初始化动作 https://cloud.google.com/dataproc/docs/concepts/configuring-clusters/init-actions在集群启动时复制 jarfiles：

#!/bin/bash
# copy-jars.sh

gsutil cp gs://foo/my-jarfile.jar /tmp/localjar.jar

如果您需要在集群部署后的某个时间动态确定将哪些 jar 文件复制到所有节点上，您可以采用这里描述的方法 https://stackoverflow.com/questions/55739463/cant-add-jars-pyspark-in-jupyter-of-google-dataproc/55859782#55859782使用初始化操作，连续监视某些 hdfs 目录以将 jarfile 复制到本地目录，然后每当您需要 jarfile 出现在所有节点上时，您只需提交一个pig fs将 JAR 文件从 GCS 放入 HDFS 的监视目录中。

通常，您不希望某些内容自动轮询 GCS 本身，因为 GCS 列表请求需要花钱，而轮询 Dataproc 集群的 HDFS 则不需要额外费用。