使用气流运行火花流作业？答案

【问题标题】：Using airflow to run spark streaming jobs?使用气流运行火花流作业？
【发布时间】：2019-02-20 11:40:39
【问题描述】：

我们的 hadoop 集群中有 Spark Batch 作业和 Spark 流式作业。

我们希望在同一个平台上安排和管理它们。

我们遇到了气流，这符合我们的需求 “用于创作、安排和监控工作流程的平台”。

我只是希望能够停止并启动 spark 流式传输作业。使用气流图和分析不是问题。

我的问题是， 除了失去一些功能（图表、分析）之外，我为什么不应该使用 Airflow 来运行 spark 流作业？

我遇到了这个问题： Can airflow be used to run a never ending task?

这说明这是可能的，而不是您不应该这样做的原因。

【问题讨论】：

标签： apache-spark streaming airflow

【解决方案1】：

@mMorozonv 看起来不错。如果流不存在，您可以让一个 DAG 启动流。然后是第二个 DAG 作为健康检查器来跟踪它的进度。如果健康检查失败，您可以再次触发第一个 DAG。

或者，您可以使用trigger 间隔once[1] 运行流。

# Load your Streaming DataFrame
sdf = spark.readStream.load(path="data/", format="json", schema=my_schema)
# Perform transformations and then write…
sdf.writeStream.trigger(once=True).start(path="/out/path", format="parquet")

这为您提供了火花流的所有相同优势，以及批处理的灵活性。

您可以简单地将流指向您的数据，此作业将检测自上次迭代以来的所有新文件（使用检查点），运行流式批处理，然后终止。您可以触发气流 DAG 的时间表，以适应您想要处理数据的任何延迟（每分钟、每小时等）。

对于低延迟要求，我不建议这样做，但它非常适合每分钟运行一次。

[1]https://databricks.com/blog/2017/05/22/running-streaming-jobs-day-10x-cost-savings.html

【讨论】：

【解决方案2】：

使用 Airflow 分支功能，我们可以使用一个 dag 来调度和监控我们的流式传输作业。 Dag 将对应用程序进行状态检查，如果应用程序未运行，dag 将提交流式作业。在另一种情况下，可以完成 dag 执行，或者您可以添加一个传感器，该传感器将在一段时间后通过警报和其他您需要的东西检查流式作业状态。

主要有两个问题：

提交流式应用程序，无需等待完成的。否则我们的算子将一直运行直到到达execution_timeout;

这个问题可以通过在cluster模式下调度流式作业来解决，spark.yarn.submit.waitAppCompletion配置参数设置为false

检查我们的流媒体运营商的状态；

我们可以使用 Yarn 检查流应用程序的状态。例如，我们可以使用命令 yarn application -list -appStates RUNNING 。如果我们的应用程序将在正在运行的应用程序列表中，我们不应该触发我们的流式作业。唯一要做的就是使流式作业名称唯一。

【讨论】：

我相信您的意思是将spark.yarn.submit.waitAppCompletion 设置为false？
@Naitree，你是对的。我已经更新了一个帖子。谢谢

【解决方案3】：

您不应该使用 Airflow 来运行 Spark Streaming 作业并没有严格的理由。事实上，您可以通过定期记录一些指标来监控您的流程

LOG.info(query.lastProgress)
LOG.info(query.status)

并在任务日志中查看它们

【讨论】：