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

Question

我们的hadoop集群中有火花批处理作业和和火花流作业。

我们希望在同一个平台上对它们进行调度和管理。

我们遇到了气流，这符合我们对“编写、调度和监控工作流的平台”的需求。

我只想能够停止和启动火花流的工作。使用气流图和剖面图不是什么问题。

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

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

这说明这是可能的，而不是为什么你不应该。

Answer 1

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

或者，您可以以trigger间隔为once1运行流。

# 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

Answer 2

使用气流分支功能，我们可以有一个dag，它将同时完成我们的流作业的调度和监视。Dag将对应用程序进行状态检查，如果应用程序没有运行，dag将提交流作业。在另一种情况下，可以完成dag的执行，或者添加一个传感器，该传感器将在一段时间后检查流作业状态，以及您需要的警报和其他内容。

有两个主要问题：

1. 提交流应用程序，而不等待它将完成。否则，我们的运营商将运行，直到它将到达execution_timeout；

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

1. 检查我们的流媒体运营商的状况；

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

Answer 3

没有严格的理由，为什么你不应该使用气流运行星火流作业。实际上，您可以通过定期记录一些指标来监视您的过程。

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

并在任务日志中看到它们