小编wal*_*r91的帖子

有人知道是否可以更改 Airflow UI 中的单个任务颜色吗？我读到可以更改运算符颜色（导入它后，您可以覆盖类方法），但是如果我有两个具有不同颜色的 PythonOperator 实例怎么办？

提前致谢！

我在使用一个简单的 Kafka 消费者程序时遇到了一些麻烦：

18/06/04 18:13:49 ERROR /log/log.txt: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:647)
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:542)
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:524)
        at com.carrefour.entequadratura.KafkaHandler.createConsumer(KafkaHandler.java:96)
        at com.carrefour.entequadratura.KafkaHandler.runConsumer(KafkaHandler.java:104)
        at com.carrefour.entequadratura.Main.main(Main.java:48)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:730)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: org.apache.kafka.common.KafkaException: java.lang.IllegalArgumentException: Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is not set
        at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:74)
        at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:60)
        at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:79)
        at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:577)
        ... 14 …

我正在尝试使用 Cloud Run 来运行容器作为 Airflow 的 DAG 的任务。

似乎没有像 CloudRunOperator 或类似的东西，我在文档中找不到任何内容（Cloud Run 和 Airflow 都可以）。

有人处理过这个问题吗？如果是，我如何使用 Cloud Run 运行容器并处理 xcom？

提前致谢！！

我对 Airflow 很陌生，我在 Xcom 和 Jinja 方面遇到了一些问题。

我必须做一些 Python 详细说明，然后将结果传递给 EmailOperator，以便将其作为电子邮件正文发送。

似乎没有关于它的文档，我发现的唯一提示是这个链接，它的格式很糟糕，不起作用，而且评论很差。我认为它必须通过 Jinja 完成，但我不知道如何......

这是我正在研究的暂定，有人可以帮助我并解释为什么这段代码是错误的以及如何修复它吗？

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2020, 11, 7),
    'email': ['ciccio.pasticcio@noreply.it'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'dg_daily_saint',
    default_args=default_args,
    schedule_interval='10 9 * * *')

task1 = PythonOperator(task_id="extract_daily_saint",
                                     python_callable=extractDailySaint,
                                     provide_context=True,
                                     dag=dag)


def html_output(**context):
    value=context['task_instance'].xcom_pull(task_ids='extract_daily_saint', key='saint')
    return "<h1>" + value + "</h1>"


EMAIL_CONTENT = """
    <b> {{ html_output(context) }}</b>
    """

mail = EmailOperator(
    task_id='mail',
    to='ciccio.pasticcio@noreply.it',
    subject='DataGovernance',
    html_content=EMAIL_CONTENT, …

有谁知道是否可以以编程方式导入/导出 DataFlow 管道（已部署或处于草稿状态）？

我们的想法是编写一个脚本来删除并创建一个 DataFusion 实例，以避免在不使用时计费。通过 gloud 命令行，可以配置 DataFusion 集群并销毁它，但自动导出和导入我的所有管道也会很有趣。

不幸的是，官方文档并没有帮助我......

谢谢！