小编Ale*_*and的帖子

我想在Airflow中创建一个条件任务,如下面的架构中所述.预期的情况如下:

• 任务1执行
• 如果任务1成功,则执行任务2a
• 否则,如果任务1失败,则执行任务2b
• 最后执行任务3

上面的所有任务都是SSHExecuteOperator.我猜我应该使用ShortCircuitOperator和/或XCom来管理这个条件,但我不知道如何实现它.你能描述一下解决方案吗？

在 CircleCI 中，我有一个执行自动化测试的构建。它使用nose2和junit插件在 XML 文件中生成测试结果。测试结果文件的完整路径为：

/project-folder/test/junit/test-result.xml

文件/project-folder/.circleci/config.yml中的我的 CircleCI 配置包含以下内容：

- store_artifacts:
    path: test/junit/test-results.xml
    destination: test-results

- store_test_results:
    path: test/junit/test-results.xml

当构建执行时，它确认测试结果和工件已正确加载：

Uploading artifacts
Uploading /home/circleci/repo/test/junit/test-results.xml to test-results
Uploaded /home/circleci/repo/test/junit/test-results.xml

Uploading test results
Archiving the following test results
  * /home/circleci/repo/test/junit/test-results.xml

Uploaded 

但是，测试结果仍然没有出现在“测试摘要”选项卡中（请参见下面的屏幕截图）。

我在这里查看了文档：

• 存储测试结果
• 自定义步骤中的元数据收集

但我看不到我错过了什么？我也不明白$CIRCLE_TEST_REPORTS上面链接中的变量指的是什么。它包含什么以及我应该如何在我的配置中使用它。

我们有一个持续集成管道，可以自动将我们的 Airflow DAG 部署到 Airflow 服务器。部署新版本的 DAG 时，其状态默认为OFF。我们希望把它ON作为由部署过程中执行的任务的一部分。

Airflow 中是否有允许打开 DAG 的命令行选项？谢谢

我有一个运行 Flask 应用程序的 Docker 容器。当 Flask 接收到 http 请求时，我想触发一个新的临时 Docker 容器的执行，该容器一旦完成它必须做的事情就会关闭。

我读过应该避免使用 Docker-in-Docker，因此这个新容器应该作为同级容器在我的主机上运行，​​而不是在 Flask 容器内运行。

使用docker-py执行此操作的解决方案是什么？

我目前有以下Dockerfile创建我的Docker映像的方法。

FROM python:3.6.6-alpine3.8

# Add dependencies for Python packages pandas, numpy and pyodbc
RUN apk add --no-cache curl gcc g++ unixodbc-dev
RUN ln -s /usr/include/locale.h /usr/include/xlocale.h

# Project files
ARG PROJECT_DIR=/srv/scripts
RUN mkdir -p $PROJECT_DIR
WORKDIR $PROJECT_DIR
COPY requirements.txt ./

# Install Python dependencies
RUN pip install --upgrade pip
RUN pip install -r requirements.txt

我想在此图中包括各种ODBC驱动程序，以便可以使用它们从容器中运行的Python程序连接到不同的数据库。

• Python程序Pyodbc用于连接数据库。
• 我需要安装的ODBC驱动程序是：
  • PostgreSQL的
  • 的MySQL
  • 女士SQL Server
  • Teradata
  • 甲骨文
  • 蜂巢
  • 黑斑羚

我想从PostgreSQL开始，认为它是最简单的，但是我在Alpine Linux Package Manager中找不到任何软件包。您知道如何安装这样的驱动程序吗？

我浏览了网络，但找不到满意的答案。是否有现有的 Docker 映像可在 Docker 容器中运行 Teradata 数据库？

到目前为止，我发现的只是这个存储库，其中包含 Hadoop Hive、Presto 等的 Docker 映像...但不包含 Teradata 数据库本身。

我需要生成一个cert.pem和key.pem文件来使用 Nginx 加密 http 请求。

在 Linux 上，我将执行以下 OpenSSL 命令：

openssl req -x509 -newkey rsa:4096 -nodes \
          -out ./nginx/config/cert.pem \
          -keyout ./nginx/config/key.pem -days 365

在 Windows 上生成此类文件的等效命令是什么？我遵循了本指南，但它只生成一个.crt和.key文件。

https://helpcenter.gsx.com/hc/en-us/articles/115015960428-How-to-Generate-a-Self-Signed-Certificate-and-Private-Key-using-OpenSSL

我想更改dag_concurrency特定Airflow DAG 的参数.似乎有一个全局dag_concurrency参数airflow.cfg但可以为不同的DAG设置不同的值吗？

我尝试在SSHExecuteOperator任务中的DAG代码中添加并发参数,但并发值仍显示DAG详细信息中的标准参数(16).

from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.hooks.ssh_hook import SSHHook
from airflow.contrib.operators.ssh_execute_operator import SSHExecuteOperator

default_args = {
  'owner': 'airflow',
  'depends_on_past': False,
  'start_date': datetime.now(),
  'email': ['exceptions@airflow.com'],
  'email_on_failure': True,
  'retries': 0
}

#server must be changed to point to the correct environment
sshHookEtl = SSHHook(conn_id='SSH__airflow@myserver')

with DAG(
  'ed_data_quality_20min-v1.6.6',
  default_args=default_args,
  schedule_interval="0,20,40 * * * *",
  dagrun_timeout=timedelta(hours=24)) as dag:
  (
    dag
    >> SSHExecuteOperator(
          task_id='run_remote_ed_data_quality_20min',
          bash_command='bash /opt/scripts/shell/EXEC_ED_DATA_QUALITY_20MIN.sh ',
          ssh_hook=sshHookEtl,
          retries=0, …

我正在使用 SQLALchemy 在 Teradata 上执行查询。我执行的查询之一是用于替换存储过程的 DDL 语句：

REPLACE PROCEDURE DEV_MIGRATION_TOOL.UNIT_TEST_NEW_STORED_PROCEDURE()
UNIT_TEST_NEW_STORED_PROCEDURE:
BEGIN
    DECLARE V_VAR VARCHAR(50);
    SELECT 'Hello World!'
    INTO :V_VAR;
END;

此 SQL 语句被分配给一个变量query，并由 SQLALchemy 使用会话执行方法执行：

def execute_sql_statement(self, query):
    """Generic method to execute a SQL statement on target environment."""
    self.target_environment.db_session.execute(query)
    self.target_environment.db_session.commit()

我:V_VAR遇到的问题是 SQLAlchemy 假设变量是一个应该随字典一起提供的参数。请参阅此处的文档：http : //docs.sqlalchemy.org/en/latest/orm/session_api.html#sqlalchemy.orm.session.Session.execute

result = session.execute("SELECT * FROM user WHERE id=:param", {"param":5})

在当前配置中，它会触发错误消息：

2018-04-18 19:09:27,874 - migration_script - INFO - 在 UAT 环境中执行对象 DDL 语句。2018-04-18 19:09:27,875 - migration_script - ERROR - (sqlalchemy.exc.InvalidRequestError) …

我有一个 python 包，如下所示：

package/\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 PyMySQL-0.7.6-py2.7.egg\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 pymysql\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 PyMySQL-0.7.x.pth  \n\xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 tests.py\n

文件夹结构无法更改，因为它来自第三方库。

.pth 文件的内容是

import sys; sys.__plen = len(sys.path)\n./PyMySQL-0.7.6-py2.7.egg\nimport sys; new=sys.path[sys.__plen:]; del sys.path[sys.__plen:]; p=getattr(sys,'__egginsert',0); sys.path[p:p]=new; sys.__egginsert = p+len(new)\n

将pymysql包含在tests.py中的最佳方法是什么