基于类型安全配置文件的Airflow schedule spark作业
是指使用Airflow调度工具来管理和执行Spark作业的一种方法。Airflow是一个开源的任务调度平台，它允许用户定义、调度和监控工作流。Spark是一个快速、通用的大数据处理引擎，可以处理大规模数据集并提供高效的数据处理能力。
在基于类型安全配置文件的Airflow schedule spark作业中，配置文件使用类型安全的格式，例如YAML或JSON，来定义Spark作业的参数和依赖关系。这种配置文件可以确保作业的参数类型正确，并提供更好的可读性和可维护性。
优势：
- 类型安全：使用类型安全的配置文件可以减少配置错误和运行时错误，提高作业的稳定性和可靠性。
- 可读性和可维护性：类型安全的配置文件提供了清晰的结构和明确的参数定义，使得配置文件易于阅读、理解和修改。
- 灵活性：通过配置文件，可以轻松地调整作业的参数和依赖关系，以满足不同的需求和场景。
应用场景： 基于类型安全配置文件的Airflow schedule spark作业适用于需要定期执行Spark作业的场景，例如数据清洗、数据转换、数据分析等。它可以帮助用户自动化地调度和执行Spark作业，并提供可视化的任务监控和报告。
推荐的腾讯云相关产品和产品介绍链接地址： 腾讯云提供了一系列与大数据处理和云计算相关的产品和服务，以下是一些推荐的产品和对应的介绍链接地址：
- 腾讯云数据工厂：https://cloud.tencent.com/product/dtfd
- 腾讯云数据仓库：https://cloud.tencent.com/product/dws
- 腾讯云弹性MapReduce：https://cloud.tencent.com/product/emr
- 腾讯云云服务器：https://cloud.tencent.com/product/cvm
- 腾讯云对象存储：https://cloud.tencent.com/product/cos
- 腾讯云人工智能平台：https://cloud.tencent.com/product/tai
请注意，以上推荐的产品和链接仅供参考，具体选择应根据实际需求和情况进行。
相关·内容
我正在使用Airflow计划向Hadoop集群提交JAR(对应于Spark应用程序的打包)。不同的Spark作业具有相同的JAR名称，但根据配置文件(我在TypeSafe配置库中使用.conf文件)启动不同的作业。我的问题是如何让Airflow解析.conf文件，以便知道要启动哪个spark作业。所以在这里，我使用spark-submit --class ( same class for all jobs) -Dconfig.file=path/to/config-file启动spark应用程序
你对这个pleaseS有什么想法吗？
浏览 13提问于2018-09-10得票数 1
我有不同的spark和airflow服务器。我没有在airflow服务器上安装spark二进制文件。我能够很好地使用SSHOperator并在集群模式下运行spark作业。我想知道从长远来看，使用SSHOperator或SparkSubmitOperator提交pyspark作业会有什么好处。任何帮助都会提前表示感谢。
浏览 9提问于2020-06-25得票数 1
有没有人可以帮我在Airflow中使用livybatchoperator，下面是我的代码…除此之外，除了spark operator之外，还有什么方法可以在气流中运行spark作业，在我的情况下，spark安装在不同的机器上。
我得到了这个错误：Getting Error in Airflow UI - "No module named 'airflow_livy'"。
from datetime import datetime, timedelta
from airflow_livy.batch import LivyBatchOperator
from a
浏览 2提问于2020-08-05得票数 0
我能够使用Spark运行BashOperator作业，但是我想使用Spark 独立模式来使用SparkSubmitOperator。
my DAG for SparkSubmitOperator和堆栈跟踪
args = {
'owner': 'airflow',
'start_date': datetime(2018, 5, 24)
}
dag = DAG('spark_job', default_args=args, schedule_interval="*/10 * * * *")
ope
浏览 0提问于2018-05-25得票数 9
回答已采纳
Spark提交命令 spark-submit --class=com.test.App /home/ubuntu/app.jar /home/ubuntu/abc.properties 用于安排spark作业的Airflow脚本 from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.operators import SparkSubmitOperator
import sys
import os
from airflow.models import Variable
fr
浏览 134提问于2019-01-04得票数 0
回答已采纳
我有一个pyspark脚本，它现在工作得很好，我想做的是，我想为每一分钟安排该作业，为此，我使用了Apache Airflow，我为airflow创建了一个.py文件，如下所示： from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
import os
from builtins import range
import airflow
from airflow.models import DAG
from
浏览 24提问于2019-04-08得票数 0
3回答
我设置了几个dags，最终以对spark集群的spark-submit命令结束。我正在使用集群模式，如果这有区别的话。总之，我的代码可以工作，但我意识到如果spark作业失败，我不一定会从Airflow UI中知道。通过集群模式触发作业，Airflow将作业交给可用的工作人员，因此airflow不知道spark作业。
我如何解决这个问题？
浏览 5提问于2017-05-18得票数 3
我正在尝试使用气流来触发本地星火独立点上的spark-examples.jar，但我总是遇到异常。当我在终端上手动提交相同的作业时，它工作如下：
spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://<HOSTNAME>:7077 \
--deploy-mode cluster \
--executor-memory 1G \
--total-executor-cores 1 \
/path/to/spark-examples_2.11-2.3.1.jar \
浏览 1提问于2018-08-01得票数 0
回答已采纳
我在GKE上运行一个GCP作曲家集群。我正在定义一个DAG来向dataproc集群提交一个作业。我读过GCP文档，它说Composer的服务帐户将被工作人员用来发送dataproc请求。
但是DataprocSubmitJobOperator报告在获取auth凭据时出错。下面是堆栈跟踪。随附作曲家个人资料。我需要建议来解决这个问题。
[2022-08-23, 16:03:25 UTC] {taskinstance.py:1448} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=harshit.bapna@dexte
浏览 9提问于2022-08-23得票数 0
1回答
我有一个码头-合成管道与容器，用于气流和火花。我想安排一个SparkSubmitOperator作业，但是它在错误java.lang.IllegalArgumentException: Too large frame: 5211883372140375593中失败了。星火应用程序只包括创建一个火花会话(我已经注释掉了所有其他内容)。当我手动运行星火应用程序(通过进入星火容器的bash并执行星火提交)时，一切都很好！此外，当我不创建一个火花会话，而只是一个SparkContext，它工作！
这是我的船坞-复合。Here：
version: '3'
x-airflow-com
浏览 3提问于2021-07-25得票数 0
回答已采纳
下面是airflow DAG代码。无论是在本地托管airflow时，还是在cloud composer上，它都能完美运行。但是，无法在Composer UI中单击DAG本身。我找到了一个类似的问题，并尝试了中链接的被接受的答案。我的问题与此类似。
import airflow
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.ope
浏览 13提问于2019-03-28得票数 1
回答已采纳
2回答
TLDR；
这是可能配置梁便携式流道与火花配置？更准确地说，可以在便携式运行程序中配置spark.driver.host吗？
动机
目前，我们已经在Kubernetes集群中实现了气流，为了使用TensorFlow扩展，我们需要使用Apache。对于我们的用例，火花将是合适的运行程序，并且由于气流和TensorFlow是在python中编码的，所以我们需要使用Apache的可移植运行程序()。
问题所在
便携式转轮在其容器内创建火花上下文，并且不为驱动程序DNS配置留出空间，使工人舱内的执行器非传染性地传递给驱动程序(作业服务器)。
设置
在beam文档之后，作业serer在与气流相同的吊
浏览 5提问于2021-02-22得票数 3
回答已采纳
1回答
为什么火花执行者还没死
、、、
这是我的设置：
运行airflow的Kubernetes集群将spark作业提交给Kubernetes集群，作业运行良好，但是容器应该在完成任务后死亡，但它们仍然挂在那里。
cluster.Dag上出现的气流设置是在气流对接器图像中烘焙的，因为不知何故，我无法将达格与s3同步。由于某些原因，cron无法运行。将spark作业提交给K8S集群，作业运行良好。，但现在它没有在执行和完成作业后死亡，它仍然存在。
这是我的SparkSubmitOperator函数
spark_submit_task = SparkSubmitOperator(
task_id='spark_submi
浏览 1提问于2019-09-17得票数 0
我在我的气流DAG中有两个任务。一个触发API调用( Http操作符)，另一个使用另一个api ( Http传感器)不断检查其状态。此DAG计划每小时& 10分钟运行一次。但有时一次执行可能需要很长时间才能完成，例如20小时。在这种情况下，前一个任务正在运行时的所有调度都不会执行。
例如，如果我在01:10完成这项工作，需要10个小时.本应运行的调度02:10、03:10、04:10、... 11:10等被跳过，只执行12:10的调度。
我正在使用本地executor。我正在使用下面的脚本运行airflow服务器和调度程序。start_server.sh
export AIRFLOW_
浏览 11提问于2019-08-09得票数 1
回答已采纳
背景:我创建了一个新的气流作业/任务DAG，其中我正在使用SparkSubmitOperator。我正在运行星火和气流在我的桌面(版本等以下)。DAG可以正常工作，直到它到达火花作业的提交部分为止。我尝试使用以下选项更改连接。无论我尝试什么，我都会在气流日志中收到以下消息。
气流识别连接并尝试使用它，但失败了。
如果我从命令提示符中提交目标DataPipelineExample.py，它将运行而不会出现问题。
问题:是什么阻止气流识别和使用火花局部的连接来执行火花提交？
Airflow.exceptions.AirflowException:无法执行:星星之交--主 -名称mySparkSub
浏览 5提问于2020-03-18得票数 1
我正在使用Gcloud Composer，试图创建一个创建DataProc集群的DAG，运行一个简单的Spark作业，然后拆卸集群。我正在尝试运行Spark PI示例作业。 我知道在调用DataProcSparkOperator时，我只能选择定义main_jar或main_class属性。当我定义main_class时，作业失败并显示以下错误： java.lang.ClassNotFoundException: org.apache.spark.examples.SparkPi
at java.net.URLClassLoader.findClass(URLClassLoader.j
浏览 37提问于2019-02-20得票数 1
回答已采纳
我们需要安排spark作业，因为我们熟悉apache-airflow，所以我们想继续使用它来创建不同的工作流。我在网上搜索，但没有找到一步一步的指南，以安排spark作业的airflow和选项，以运行它们在不同的服务器运行主机。
对此问题的回答将受到高度赞赏。提前谢谢。
浏览 4提问于2018-11-17得票数 8
回答已采纳
1回答
我在airflow中有以下任务，它的工作非常有魅力：
t = SparkSubmitOperator(
task_id = 'some_id',
application = '/path/to/app.py',
name = 'airflow-spark',
conf = {
'spark.hadoop.fs.s3a.endpoint': 'https://some.url.com/',
'spark.hadoop.fs.s3a.access.
浏览 3提问于2020-11-23得票数 0
回答已采纳
1回答
我的Airflow per服务器已启动并运行，因为其他作业正在按计划运行。我添加了一个新的DAG，每5分钟执行一次。添加后，我第一次手动运行它，它完成了。然而，在那之后，它不会每隔5分钟再次运行一次。 Dag代码如下 from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
current_date = datetime.now()
default_args = {
"owner":
浏览 12提问于2019-06-11得票数 0
我需要在AWS EMR中运行结构化流spark作业。作为弹性需求，如果spark作业由于某些原因而失败，我们希望spark作业可以在EMR中重新创建。类似于ECS中的任务编排，如果健康检查失败，可以重启任务。然而，EMR更多是一个计算引擎，而不是编排系统。 我正在寻找一些大数据工作流程编排工具，例如Airflow。但是，它不能支持DAG中的循环。如何实现如下所示的一些功能？ step_adder (EmrAddStepsOperator) >> step_checker (EmrStepSensor) >> step_adder (EmrAddStepsOperato
浏览 44提问于2020-07-13得票数 1
回答已采纳
我想为conn_id生成spark_default。我正在k8s上运行我的k8s，我想使用火花主程序动态地生成conn_id，这是运行在同一个名称空间中的另一个容器。
是否有办法动态生成conn_id :类似于：
环境变量
或者使用SparkSubmitOperator本身编写和生成conn_id
下面是我的dag代码：
from airflow import DAG
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import date
浏览 1提问于2019-05-23得票数 0
在气流中，我使用KubernetesExecutor作为执行器。我的DAG代码
from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.spark_kubernetes import SparkKubernetesOperator
from airflow.providers.cncf.kubernetes.sensors.spark_kubernetes import SparkKubernetesSensor
dag
浏览 32提问于2022-01-29得票数 1
气流2.3.3
我有两个TaskGroups的达格。每个TaskGroup有两个任务：
t1: SparkKubernetesOperator >> t2: SparkKubernetesSensor
t1使用spark部署yaml文件向kubernetes集群提交spark作业。立即进入深绿色成功状态。
t2监视t1的执行。如果火花作业正在运行，则需要10分钟才能完成，然后t2进入成功状态。
我有这种情况，然后提交火花作业出错: UnknownHostException，这是我想重试的时候，但是我想重试整个TaskGroup，而不仅仅是t2。
我现在不可能重试整个TaskGrou
浏览 20提问于2022-09-01得票数 0
3回答
大家好，地球上的人们！我正在使用Airflow来安排和运行Spark任务。到目前为止，我发现的是Airflow可以管理的python DAG。
DAG示例：
spark_count_lines.py
import logging
from airflow import DAG
from airflow.operators import PythonOperator
from datetime import datetime
args = {
'owner': 'airflow'
, 'start_date': datetime(20
浏览 0提问于2016-10-03得票数 41
回答已采纳
我正在使用Google dataproc提交spark作业，并使用google cloud composer来安排它们。不幸的是，我正面临着困难。
我依靠.conf文件(类型安全配置文件)将参数传递给我的spark作业。
我对airflow数据过程使用了以下python代码：
t3 = dataproc_operator.DataProcSparkOperator(
task_id ='execute_spark_job_cluster_test',
dataproc_spark_jars='gs://snapshots/jars/pubsub-assemb
浏览 0提问于2018-09-15得票数 3
我正在使用airflow bash操作符来运行spark-submit作业。spark作业需要一些参数。最后一个参数是一个字符串，可以使用"，“和"|”split将其转换为嵌套数组。提交命令示例：
spark-submit --verbose --master yarn --deploy-mode cluster --queue some_queue job.jar <dt> <s3bucket> <xx:yyy,zzz|ss:aaa,bbb>
奇怪的是，当spark任务完成时，它会输出：
final status: SUCCEEDED
Co
浏览 12提问于2018-08-31得票数 0
我对气流很陌生，我试图在码头集装箱中布置一个喷火作业，这是我的达格，
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import datetime, timedelta
spa
浏览 0提问于2021-07-09得票数 1
我的Ariflow脚本有几个任务，使用python操作符启动一个使用boto3的粘合作业。即使流程依赖于Ariflow中的每个任务。连续的任务不会等到前一个任务成功完成。 一旦调用了前一个粘合作业任务，就会调用下一个粘合作业任务。最后，尽管气流看起来成功完成，但胶水作业仍然运行了几分钟。 from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
from airflow
浏览 29提问于2019-05-28得票数 6
1回答
我有一个码头-撰写文件，在其中定义的服务气流，火花，postgreSQL和雷迪斯。当我执行对接-撰写和打开气流UI，我尝试添加一个火花连接类型，以便我可以运行火花作业内部气流在码头。但是，正如您可以看到的那样，Spark作为连接类型不可见：
这是我的停靠-撰写文件：
# Feel free to modify this file to suit your needs.
---
version: '3'
x-airflow-common:
&airflow-common
# In order to add custom dependencies or u
浏览 11提问于2022-01-31得票数 0
1回答
我们正在使用Airflow来计划我们在电子病历上的作业，目前我们想使用apache Livy通过Airflow提交Spark作业，我需要更多的指导如下:哪种Airflow-Livy操作符我们应该用于python、3+、pyspark和scala作业。我看过以下内容：和
想知道更多关于稳定的AirflowLivy操作符，任何人在生产中使用，可能是在亚马逊网络服务堆栈。
还提供了集成的逐步安装指南。
浏览 4提问于2020-04-13得票数 0
我有火花和气流集群，我想发送一个火花作业从气流容器到火花容器。但我对气流是新的，我不知道我需要执行哪种配置。我在插件文件夹下复制了spark_submit_operator.py。
from airflow import DAG
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import datetime, timedelta
args = {
'owner': 'airflow',
浏览 0提问于2018-09-20得票数 4
2回答
我使用的是下面的停靠-组合映像，我从以下位置获得了这个映像：
version: "3"
x-airflow-common: &airflow-common
image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.0.0-python3.7}
environment: &airflow-common-env
AIRFLOW__CORE__EXECUTOR: CeleryExecutor
AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airfl
浏览 26提问于2021-06-05得票数 17
回答已采纳
我正在使用Apache批处理POST方法提交Spark作业。
这个HTTP请求是使用AirFlow发送的。提交作业后，我使用批处理Id跟踪状态。
我想在气流日志上显示驱动程序(客户端日志)日志，以避免AirFLow和Apache /Resource访问多个地方。
使用Apache可以这样做吗？
浏览 4提问于2019-01-20得票数 4
回答已采纳
我对docker还是个新手，正在尝试用airflow和pyspark来运行docker-compose文件。以下是我到目前为止所掌握的： version: '3.7'
services:
master:
image: gettyimages/spark
command: bin/spark-class org.apache.spark.deploy.master.Master -h master
hostname: master
environment:
MASTER: spark://master:70
浏览 34提问于2020-10-23得票数 2
2回答
我有一个火花和气流集群，这是建立与码头群。气流容器不能如我所期望的那样包含火花提交。
我正在使用github中存在的以下图像
火花:大数据-欧洲/码头-hadoop-火花-工作台
气流:puckel/码头-气流(CeleryExecutor)
我准备了一个.py文件并将其添加到dags文件夹下。
from airflow import DAG
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import datetime, timedelta
arg
浏览 0提问于2018-09-24得票数 2
我们有一个Airflow DAG，它涉及在Dataproc上运行pyspark作业。在作业期间，我们需要一个jdbc驱动程序，我通常会将其传递给dataproc提交命令：
gcloud dataproc jobs submit pyspark \
--cluster my-cluster \
--properties spark.jars.packages=mysql:mysql-connector-java:6.0.6 \
--py-files ...
但是我怎么才能用Airflow的DataProcPySparkOperator来做呢？
现在，我们将这个库添加到集群本身：
gcloud d
浏览 2提问于2017-11-14得票数 0
我正在考虑在我的工作场所建立一个混合云设置。现在，我们在AWS上的Spark集群上运行了机器学习作业。我想要做的是能够从GCP上运行的Airflow作业连接到该集群，以便根据需要调用各种作业并执行其他任务。
不过，我不知道该怎么做。我可以为Spark集群创建一个公共DNS，这将使它很容易通过互联网从GCP连接，但我认为这对于安全性来说并不理想。
我是否需要在GCP项目(即GCP VPC)和AWS VPC之间设置VPN？如果是这样，我如何做到这一点？如果有办法使用普通的GCP和AWS配置，我不想使用第三方产品。
基本上，我认为我需要的是GCP上的工作能够看到Spark集群的内部DNS，并能够使用
浏览 0提问于2020-01-12得票数 1
我正试图在我的笔记本电脑上运行Airflow测试ETL pyspark作业。我已经在同一个虚拟环境中安装了Spark和Airflow。值得一提的是，我可以从虚拟环境中成功运行作业，但它在Airflow中失败。从Airflow日志中，我可以看到ETL任务失败，因为它在尝试执行时在临时目录中查找dag文件。下面是错误："python: can't open file '/tmp/airflowtmpd9uwr02_/spark_etl_1.py'“(完整日志如下)。
> AIRFLOW_CTX_DAG_OWNER=florian
AIRFLOW_CTX_DAG
浏览 65提问于2021-02-24得票数 1
嗨，我还是个新手。我想知道气流是如何连接到Spark的？我有一个服务器，它使用不同的帐户(或配置文件)来访问Spark集群。我想知道Airflow是否有固定的轮廓来连接火花？或者它遵循用户配置文件？
另外，是否可以在Airflow中为用户设置不同的访问权限？例如，在Spark (或hdfs)中有权访问位置A的用户A只能运行与位置A相关的任务(因此无法从位置B运行任何作业)。
提前谢谢。
浏览 7提问于2017-05-19得票数 0
Apache Airflow版本: v2.1.1
Kubernetes版本(如果使用kubectl )(使用kubectl版本)：-客户端版本:version.Info{主要：“1”，次要：“21”，GitVersion:"v1.21.2"，GitCommit:"092fbfbf53427de67cac1e9fa54aaa09a28371d7"，GitTreeState:"clean"，BuildDate:"2021-06-16T12:52:14Z"，GoVersion:"go1.16.5"，编译器：“gc”，平台
浏览 2提问于2021-07-14得票数 3
我有一个气流DAG，我用它来提交一个spark作业，为此我使用了SparkSubmitOperator。在DAG中，我必须指定需要运行的应用程序JAR。目前，它被硬编码为spark-job-1.0.jar，如下所示：
from airflow import DAG
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import datetime, timedelta
args = {
'owner': 'joe'
浏览 12提问于2020-05-23得票数 0
回答已采纳
我对气流很陌生，我需要在用例中读取传入的json信任信息，然后根据所读取的信任值构造一个字符串，这个字符串将用作我们正在GCP数据过程中创建的集群的名称。
例:输入Json到dag
{ "x“："data"，"y"："engg"，"z”："usecase“}
我希望集群名是“engg”，并包含下面的代码，但是我得到了一个错误，就像它无法识别"dag_run“一样。任何帮助都将不胜感激。
from datetime import datetime
from airflow import DAG
from airflo
浏览 6提问于2022-06-09得票数 2
回答已采纳
TLDR；
在Kubernetes中连接2个pods是可能的，因为它们在相同的局域网中，所有的端口都打开了？
动机
目前，我们已经在Kubernetes集群中实现了airflow，并且为了使用TensorFlow扩展，我们需要使用Apache。对于我们的用例，Spark将是要使用的合适的runner，因为airflow和TensorFlow是用python编写的，所以我们需要使用Apache Beam的Portable Runner ()。
问题所在
气流pod和作业服务器pod之间的通信导致传输错误(可能是因为作业服务器使用的一些随机端口)。
设置
为了遵循良好的隔离实践并模仿Kuberne
浏览 43提问于2021-02-23得票数 2
我正在使用我的DAG中的EMR CreateJobFlow、AddSteps、StepSensor和TerminateJobFlow操作符来启动EMR集群，添加步骤(2个spark app和dist-cp)，并在所有步骤完成或1个步骤失败时终止。当我有一个2步的DAG (第一步是Spark应用程序，第二步是dist-cp)时，我能够做到这一点，但是，当我有两个spark应用程序时，集群成功地运行了第一步，并且在没有移动到第二步和第三步的情况下终止。
通过一些挖掘，我可以看到气流“戳”着脚步，看看它们是否还在运行。在这种情况下，它似乎认为只有当一个步骤完成时它才是“成功的”。
我的spark应用
浏览 0提问于2018-10-29得票数 2
我正在尝试在ubuntu EC2实例的后台运行Airflow调度器和EC2服务器。每当我启动启动作业时，airflow-webserver作业直接转到停止/等待。下面是我保存在/etc/init中的.conf文件。(我使用anaconda安装了airflow)
description "Airflow webserver daemon"
start on started networking
stop on (deconfiguring-networking or runlevel [016])
respawn
respawn limit 5 30
setuid unbu
浏览 5提问于2018-07-24得票数 0
我正在尝试将一个通过yaml文件配置的spark-submit作业转换为一个airflow dag，并通过airflow运行它。当我尝试装载卷并加载配置映射时，作业失败。在yaml中，卷挂载的代码块如下所示
volumeMounts:
- mountPath: /etc/jtp
name: config-vol
要加载的配置映射如下所示
volumes:
- configMap:
name: qa-jtp-llt
name: config-vol
现在，当我尝试将相同的东西转换为我的dag时，它会转换为如下所示
from kubernet
浏览 3提问于2020-10-14得票数 1
我对气流很陌生。然而，我遇到了与新手在处理日程间隔时所面临的相同的时间和间隔问题。因此，我想尝试通过cli从外部触发DAG。这可以通过简单地进入控制台并输入example：airflow trigger_dag tutorial (使用airflow docker图像: 1.10.9)来完成。 接下来，我想看看相同的命令是否适用于常规的cron作业，就像我想要触发的cron作业时间一样。因此，我创建了一个类似以下内容的cron作业：* * * * * airflow trigger_dag tutorial 但是，这现在不会触发DAG。在其他一些实验中，我可以通过shell脚本中的相同命令手
浏览 21提问于2021-07-31得票数 0
1回答
我正试着安排一个每月的气流工作。我把开始日期
'start_date':datetime(2020,9,23),
这是前一个月的日期(今天的日期)；因为“start_date+schedule_interval”规则。我的时间表间隔如下：
schedule_interval="20 9 23 * *"
按照这种逻辑，作业应该运行在2020/23/10 9:23世界协调时。但我不知道为什么它没有运行，甚至没有创建一个实例。我做的一切都是正确的，把开始日期保持在一个月前，甚至尝试了catchup= True。但没什么用。
如果我试着把日程安排保持在每天的水平上，作
浏览 1提问于2020-10-23得票数 0
回答已采纳
我想给通过Airflow DataprocSubmitJobOperator在谷歌云上提交的spark作业提供一个自定义作业id。 通过API，我们可以使用--id参数来做这件事，你知道我们如何通过这个运算符给出同样的结果吗？
浏览 24提问于2021-10-05得票数 2
回答已采纳
在DAG中是否有自定义电子邮件和发送任何任务失败的选项。有一个类似‘email _on_failure’的选项: True，但它不提供动态添加内容到电子邮件主题或正文的选项。
我的DAG将如下所示
import airflow
from airflow import DAG
from airflow.contrib.operators.databricks_operator import DatabricksSubmitRunOperator
from airflow.operators.email_operator import EmailOperator
from airflow.ope
浏览 2提问于2018-08-07得票数 7
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
