小编Fel*_* FB的帖子

我正在尝试通过 Azure AKS 使用 K8S。

但是当执行一个简单的命令时，例如：kubectl createnamespaceairflow

我收到以下错误消息：

来自服务器的错误（禁止）：禁止命名空间：用户“xxx”无法在集群范围内的 API 组“”中创建资源“命名空间”

我已经命令 az aks get-credentials 连接到集群，然后尝试创建命名空间，但没有成功。

我刚刚运行了一个气流 DAG。当我看到气流上次运行日期时，它显示最后但最后一次运行日期。当我将鼠标悬停在“i”图标上时，它引起了我的注意，它显示了正确的日期。有什么办法可以解决这个问题吗？听起来像是无稽之谈，但我最终用它来对我的数据进行质量检查。

我在我的机器上安装了 Python、Docker 并试图导入， from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator但是当我连接 docker 时，我收到该模块不存在的消息。我已经做了pip install apache-airflow[kubernetes]，我仍然有同样的错误。是否有特定的机器位置我应该检查库是否已实际安装？我能做些什么来解决这个问题？

from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.operators.dummy_operator import DummyOperator
import logging
import os
from airflow.utils.helpers import parse_template_string

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime.utcnow(),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'kubernetes_sample', default_args=default_args, schedule_interval=timedelta(minutes=10))


start = DummyOperator(task_id='run_this_first', dag=dag)

passing = KubernetesPodOperator(namespace='default',
                          image="Python:3.6",
                          cmds=["Python","-c"],
                          arguments=["print('hello world')"],
                          labels={"foo": "bar"},
                          name="passing-test",
                          task_id="passing-task", …

我正在尝试逐步按照此步骤上传 Kubernetes ( https://github.com/EamonKeane/airflow-GKE-k8sExecutor-helm ) 中的气流，但在执行的这一部分中，我遇到了如下问题：

对该主题的研究没有发现任何可以解决我的问题的东西，有没有人对该怎么做有任何建议？

SQL_ALCHEMY_CONN=postgresql+psycopg2://$AIRFLOW_DB_USER:$AIRFLOW_DB_USER_PASSWORD@$KUBERNETES_POSTGRES_CLOUDSQLPROXY_SERVICE:$KUBERNETES_POSTGRES_CLOUDSQLPROXY_PORT/$AIRFLOW_DB_NAME

echo $SQL_ALCHEMY_CONN > /secrets/airflow/sql_alchemy_conn
# Create the fernet key which is needed to decrypt database the database
FERNET_KEY=$(dd if=/dev/urandom bs=32 count=1 2>/dev/null | openssl base64)
echo $FERNET_KEY > /secrets/airflow/fernet-key

kubectl create secret generic airflow \
    --from-file=fernet-key=/secrets/airflow/fernet-key \
    --from-file=sql_alchemy_conn=/secrets/airflow/sql_alchemy_conn

无法连接到服务器：执行访问令牌命令“/google/google-cloud-sdk/bin/gcloud config config-helper --format=json”时出错：err=exit status 1 output= stderr=ERROR: gcloud crashed ( BadStatusLine): '' 如果您想报告此问题，请运行以下命令：gcloud feedback 要检查 gcloud 是否存在常见问题，请运行以下命令：gcloud info --run-diagnostics

我知道已经有一个气流函数可以将文件从 Cloud Storage 传递到 Big Query，正如我所做的那样，我在脚本内部与 GCP 建立了连接，就像在没有气流的情况下一样，我调用了 PythonOperator 来调用该函数我在脚本中配置为读取 Cloud Storage 并将文件中的数据插入到 Big Query，但是我收到错误消息：“得到了一个意外的关键字参数‘dag’”

这似乎是一件很容易解决的事情，但我真的不知道这意味着什么，因为我将 DAG 属性放在 PythonOperator 中

import json
import decimal
import airflow
from airflow import DAG
from airflow.contrib.operators.mysql_to_gcs import MySqlToGoogleCloudStorageOperator
from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
from datetime import datetime, timedelta
from airflow.operators.bash_operator import BashOperator
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from airflow.contrib.hooks.gcs_hook import GoogleCloudStorageHook
from airflow.hooks.mssql_hook import MsSqlHook
from tempfile import NamedTemporaryFile
import pymssql  
import logging
import os
# import cloudstorage …

我通过 docker 在 GCP 上的 VM 内运行气流（1.10.1）。已经更改了我的 VM 的本地时间，并且配置 (airflow.cfg) 还设置了我所在国家/地区（美国/圣保罗）的 default_zone 但它仍然在主屏幕上以 UTC 时间继续，因此处理也在 UTC 中完成。你还能做点别的吗？

大家好，这次我遇到了一个新的数组分组错误，我将与您分享我正在咨询的表的架构，以便您告诉我解决方案。我尝试使用 ARRAY_TO_STRING 但在这种情况下它没有成功......

SELECT
    individual_details.gender  AS gender,
    COUNT(DISTINCT profile.owner_id ) AS profile_count_distinct
FROM dataset.profile  AS profile
LEFT JOIN UNNEST(profile.individual_details) as individual_details
GROUP BY 1
ORDER BY 2 DESC

UNNEST 中引用的值必须是数组。UNNEST 在 [5:18] 处包含类型为 STRUCT 的表达式

我正在尝试制定每周、每月的气流计划，但不起作用。有人可以报告可能发生的情况吗？如果我每周、每月进行安排，它就会保持静止，就好像它被关闭一样。没有错误信息，只是不执行。我发送了一个代码示例来演示我如何安排...还有其他方法可以进行此安排吗？

import airflow
import os
import six
import time
from datetime import datetime, timedelta
from airflow import DAG
from airflow import AirflowException
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from airflow.operators.slack_operator import SlackAPIPostOperator

default_args = {
    'owner': 'bexs-data',
    'start_date': airflow.utils.dates.days_ago(0),
    'depends_on_past': False,
    'email': ['airflow@apache.org'],
    'email_on_failure': False,
    'email_on_retry': False,
    'depends_on_past': False,
    # If a task fails, retry it once after waiting
    # at least 5 minutes
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'on_failure_callback': slack_msg
}

dag = DAG(
    dag_id=nm_dag,
    default_args=default_args,
    schedule_interval='51 18 …

是否可以使用 Python 将 Cloud Storage 中已有的数据导入到 bigquery 中的临时表？我可以在 Python 中创建 BigQuery 临时表并将数据插入其中吗？