小编Mr.*_*ent的帖子

我正在尝试使用systemd在ubuntu服务器上运行气流.我已经按照快速入门指南和气流文档中的教程进行了操作,并设法安装了气流并使用以下命令成功运行它:

airflow webserver -p 8080

在使用配置文件安装systemd和大量试验和错误后,我设法使用该命令运行气流

sudo systemctl start airflow

气流持续运行一周,直到今天我用命令重新启动它

sudo systemctl restart airflow

sudo systemctl status airflow现在运行给我以下两条消息之一:

? airflow.service - Airflow webserver daemon
 Loaded: loaded (/lib/systemd/system/airflow.service; enabled; vendor preset: enabled)
 Active: activating (auto-restart) (Result: exit-code) since Wed 2018-09-12 09:23:01 UTC; 1s ago
Process: 3115 ExecStart=/opt/miniconda3/bin/airflow webserver -p 8080 --pid /home/user/airflow/airflow-webserver.pid --daemon (code=exited, status=1/FAILURE)
Main PID: 3115 (code=exited, status=1/FAILURE)

Sep 12 09:23:01 server-service systemd[1]: airflow.service: Main process exited, code=exited, status=1/FAILURE
Sep 12 …

我正在使用气流来编排一些python脚本.我有一个"主"dag,从中运行几个子标记.我的主要dag应该按照以下概述运行:

我已经设法通过使用以下行在我的主要dag中获得此结构:

etl_internal_sub_dag1 >> etl_internal_sub_dag2 >> etl_internal_sub_dag3
etl_internal_sub_dag3 >> etl_adzuna_sub_dag
etl_internal_sub_dag3 >> etl_adwords_sub_dag
etl_internal_sub_dag3 >> etl_facebook_sub_dag
etl_internal_sub_dag3 >> etl_pagespeed_sub_dag

etl_adzuna_sub_dag >> etl_combine_sub_dag
etl_adwords_sub_dag >> etl_combine_sub_dag
etl_facebook_sub_dag >> etl_combine_sub_dag
etl_pagespeed_sub_dag >> etl_combine_sub_dag

我想风做的是先运行etl_internal_sub_dag1然后etl_internal_sub_dag2,然后etl_internal_sub_dag3.当etl_internal_sub_dag3完成我想etl_adzuna_sub_dag,etl_adwords_sub_dag,etl_facebook_sub_dag,和etl_pagespeed_sub_dag并行运行.最后,当最后四个脚本完成后,我想要etl_combine_sub_dag运行.

然而,当我经营的主要DAG, ,etl_adzuna_sub_dag,etl_adwords_sub_dag,etl_facebook_sub_dag和etl_pagespeed_sub_dag 正在运行一个接一个,而不是并行.

问:如何确保脚本etl_adzuna_sub_dag,etl_adwords_sub_dag,etl_facebook_sub_dag,和etl_pagespeed_sub_dag并行运行？

编辑:我default_args和DAG这个样子:

default_args = {
    'owner': …

我正在做关于机器学习的本教程,其中使用了以下代码:

import pandas as pd
from sklearn.model_selection import train_test_split
df = pd.read_csv('breast-cancer-wisconsin.data.csv')
df.replace('?', -99999, inplace = True)
df.drop(['id'], 1, inplace = True)
X = np.array(df.drop(['class'], 1))
y = np.array(df['class'])

X_train, X_test, y_test, y_train = train_test_split(X, y)

以下是csv文件中的示例:

id,clump_thickness,unif_cell_size,unif_cell_shape, marg_adhesion,
single_epith_cell_size,bare_nuclei,bland_chrom,norm_nucleoli, mitoses,class
    1000025,5,1,1,1,2,1,3,1,1,2
    1002945,5,4,4,5,7,10,3,2,1,2
    1015425,3,1,1,1,2,2,3,1,1,2
    1016277,6,8,8,1,3,4,3,7,1,2
    1017023,4,1,1,3,2,1,3,1,1,2
    1017122,8,10,10,8,7,10,9,7,1,4
    1018099,1,1,1,1,2,10,3,1,1,2
    1018561,2,1,2,1,2,1,3,1,1,2
    1033078,2,1,1,1,2,1,1,1,5,2
    1033078,4,2,1,1,2,1,2,1,1,2
    1035283,1,1,1,1,1,1,3,1,1,2
    1036172,2,1,1,1,2,1,2,1,1,2
    1041801,5,3,3,3,2,3,4,4,1,4
    1043999,1,1,1,1,2,3,3,1,1,2
    1044572,8,7,5,10,7,9,5,5,4,4
    1047630,7,4,6,4,6,1,4,3,1,4
    1048672,4,1,1,1,2,1,2,1,1,2
    1049815,4,1,1,1,2,1,3,1,1,2
    1050670,10,7,7,6,4,10,4,1,2,4
    1050718,6,1,1,1,2,1,3,1,1,2
    1054590,7,3,2,10,5,10,5,4,4,4
    1054593,10,5,5,3,6,7,7,10,1,4
    1056784,3,1,1,1,2,1,2,1,1,2
    1057013,8,4,5,1,2,?,7,3,1,4
    1059552,1,1,1,1,2,1,3,1,1,2
    1065726,5,2,3,4,2,7,3,6,1,4
    1066373,3,2,1,1,1,1,2,1,1,2

在查看结果时,sklearn.model_selection.train_test_split我发现了一些奇怪的东西(至少对我而言).如果我跑

    print(type(y_test[0]))
    print()
    print(type(X_train[:,1][0]))

我得到以下输出:

<class 'numpy.int64'>
<class …

我正在使用Dash使用python构建仪表板。我有以下代码：

import dash
import dash_core_components as dcc
import dash_table_experiments as dte
import dash_html_components as html
from datetime import datetime as dt

app = dash.Dash()

app.layout = html.Div([
    dcc.Location(id = 'url', refresh=False),
    html.Div(id = 'page-content')
])
app.config['suppress_callback_exceptions'] = True

def get_index():
    index = html.Div([
        dcc.Link('Live Data', href='/live')
    ], className = 'row')
    return index

live = html.Div([
    html.H1('Table '),
    dcc.DatePickerRange(
            id = 'selected-period',
            initial_visible_month=dt(2018, 4, 5),
            start_date = dt(2018, 4, 22),
            end_date=dt(2018, 4, 25)
    ),
    # dte.DataTable(
    #     rows= [{}], …

我正在尝试让气流工作以更好地编排 etl 过程。当我对 dags 文件夹中的 dag 进行更改时，我经常需要重新启动调度程序

airflow scheduler

在更改在 UI 中可见之前。我想将调度程序作为守护进程运行

airflow scheduler -D

但我们尝试这样做，我收到一条消息说

[2018-10-17 14:13:54,769] {jobs.py:580} ERROR - 
Cannot use more than 1 thread when using sqlite. Setting max_threads to 1

我认为出现此错误是因为调度程序已经作为守护程序运行。但是，当我尝试找出调度程序正在运行的位置时

lsof -i

我没有得到任何结果。

问题：为什么我无法使用airflow scheduler -D. 为什么调度程序重新启动airflow webserver？如何成功终止阻止我运行的进程airflow scheduler -D？