[jira] [Created] (AIRFLOW-357) how should I use the right owner task in airflow?
wei.he created AIRFLOW-357: -- Summary: how should I use the right owner task in airflow? Key: AIRFLOW-357 URL: https://issues.apache.org/jira/browse/AIRFLOW-357 Project: Apache Airflow Issue Type: Bug Affects Versions: Airflow 1.7.1 Reporter: wei.he I dont understand the "owner" in airflow. the comment of ower is "the owner of the task, using the unix username is recommended". I wrote some the following code. Default_args = { 'owner': 'max', 'depends_on_past': False, 'start_date': datetime(2016, 7, 14), 'email': ['m...@test.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), dag = DAG('dmp-annalect', default_args=default_args, schedule_interval='30 0 * * *') task1_pigjob_basedata = """ {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & """.format(local_dir=WORKSPACE) task1_pigjob_basedata = BashOperator( task_id='task1_pigjob_basedata_impclk',owner='max', bash_command=pigjob_basedata_impclk, dag=dag) I used the command "airflow test dagid taskid 2016-07-20" , But I got a error, ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, I thought that my job ran with "max" user, but apperently , ran test using 'airflow' user . I hope if I run my task using 'max' user, how should I do. Thanks -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (AIRFLOW-357) how should I use the right owner task in airflow?
[ https://issues.apache.org/jira/browse/AIRFLOW-357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] wei.he updated AIRFLOW-357: --- Description: I dont understand the "owner" in airflow. the comment of ower is "the owner of the task, using the unix username is recommended". I wrote some the following code. Default_args = { 'owner': 'max', 'depends_on_past': False, 'start_date': datetime(2016, 7, 14), 'email': ['m...@test.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), dag = DAG('dmp-annalect', default_args=default_args, schedule_interval='30 0 * * *') task1_pigjob_basedata = """ {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & """.format(local_dir=WORKSPACE) task1_pigjob_basedata = BashOperator( task_id='task1_pigjob_basedata_impclk',owner='max', bash_command=pigjob_basedata_impclk, dag=dag) I used the command "airflow test dagid taskid 2016-07-20" , But I got a error, ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, I thought that my job ran with "max" user, but apperently , ran test using 'airflow' user . I hope if I run my task using 'max' user, how should I do. Thanks was: I dont understand the "owner" in airflow. the comment of ower is "the owner of the task, using the unix username is recommended". I wrote some the following code. Default_args = { 'owner': 'max', 'depends_on_past': False, 'start_date': datetime(2016, 7, 14), 'email': ['m...@test.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), dag = DAG('dmp-annalect', default_args=default_args, schedule_interval='30 0 * * *') task1_pigjob_basedata = """ {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & """.format(local_dir=WORKSPACE) task1_pigjob_basedata = BashOperator( task_id='task1_pigjob_basedata_impclk',owner='max', bash_command=pigjob_basedata_impclk, dag=dag) I used the command "airflow test dagid taskid 2016-07-20" , But I got a error, ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, I thought that my job ran with "max" user, but apperently , ran test using 'airflow' user . I hope if I run my task using 'max' user, how should I do. Thanks > how should I use the right owner task in airflow? > - > > Key: AIRFLOW-357 > URL: https://issues.apache.org/jira/browse/AIRFLOW-357 > Project: Apache Airflow > Issue Type: Bug >Affects Versions: Airflow 1.7.1 >Reporter: wei.he > > I dont understand the "owner" in airflow. the comment of ower is "the owner > of the task, using the unix username is recommended". I wrote some the > following code. > Default_args = { > 'owner': 'max', > 'depends_on_past': False, > 'start_date': datetime(2016, 7, 14), > 'email': ['m...@test.com'], > 'email_on_failure': False, > 'email_on_retry': False, > 'retries': 1, > 'retry_delay': timedelta(minutes=5), > dag = DAG('dmp-annalect', default_args=default_args, > schedule_interval='30 0 * * *') > task1_pigjob_basedata = """ > {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & > """.format(local_dir=WORKSPACE) > task1_pigjob_basedata = BashOperator( > task_id='task1_pigjob_basedata_impclk',owner='max', > bash_command=pigjob_basedata_impclk, > dag=dag) > I used the command "airflow test dagid taskid 2016-07-20" , But I got a > error, ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, > > I thought that my job ran with "max" user, but apperently , ran test using > 'airflow' user . > I hope if I run my task using 'max' user, how should I do. > Thanks -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (AIRFLOW-357) how should I use the right owner task in airflow?
[ https://issues.apache.org/jira/browse/AIRFLOW-357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] wei.he updated AIRFLOW-357: --- Description: I dont understand the "owner" in airflow. the comment of ower is "the owner of the task, using the unix username is recommended". I wrote some the following code. Default_args = { 'owner': 'max', 'depends_on_past': False, 'start_date': datetime(2016, 7, 14), 'email': ['m...@test.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), dag = DAG('dmp-annalect', default_args=default_args, schedule_interval='30 0 * * *') task1_pigjob_basedata = """ {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & """.format(local_dir=WORKSPACE) task1_pigjob_basedata = BashOperator( task_id='task1_pigjob_basedata_impclk',owner='max', bash_command=pigjob_basedata_impclk, dag=dag) I used the command "airflow test dagid taskid 2016-07-20" , But I got a error, ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, I thought that my job ran with "max" user, but apperently , ran test using 'airflow' user . I hope if I run my task using 'max' user, how should I do. Thanks was: I dont understand the "owner" in airflow. the comment of ower is "the owner of the task, using the unix username is recommended". I wrote some the following code. Default_args = { 'owner': 'max', 'depends_on_past': False, 'start_date': datetime(2016, 7, 14), 'email': ['m...@test.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), dag = DAG('dmp-annalect', default_args=default_args, schedule_interval='30 0 * * *') task1_pigjob_basedata = """ {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & """.format(local_dir=WORKSPACE) task1_pigjob_basedata = BashOperator( task_id='task1_pigjob_basedata_impclk',owner='max', bash_command=pigjob_basedata_impclk, dag=dag) I used the command "airflow test dagid taskid 2016-07-20" , But I got a error, ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, I thought that my job ran with "max" user, but apperently , ran test using 'airflow' user . I hope if I run my task using 'max' user, how should I do. Thanks > how should I use the right owner task in airflow? > - > > Key: AIRFLOW-357 > URL: https://issues.apache.org/jira/browse/AIRFLOW-357 > Project: Apache Airflow > Issue Type: Bug >Affects Versions: Airflow 1.7.1 >Reporter: wei.he > > I dont understand the "owner" in airflow. the comment of ower is "the owner > of the task, using the unix username is recommended". I wrote some the > following code. > Default_args = { > 'owner': 'max', > 'depends_on_past': False, > 'start_date': datetime(2016, 7, 14), > 'email': ['m...@test.com'], > 'email_on_failure': False, > 'email_on_retry': False, > 'retries': 1, > 'retry_delay': timedelta(minutes=5), > dag = DAG('dmp-annalect', default_args=default_args, > schedule_interval='30 0 * * *') > task1_pigjob_basedata = """ > {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & > """.format(local_dir=WORKSPACE) > task1_pigjob_basedata = BashOperator( > task_id='task1_pigjob_basedata_impclk',owner='max', > bash_command=pigjob_basedata_impclk, > dag=dag) > I used the command "airflow test dagid taskid 2016-07-20" , > But I got a error, > ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, > I thought that my job ran with "max" user, but apperently , ran test using > 'airflow' user . > I hope if I run my task using 'max' user, how should I do. > Thanks -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (AIRFLOW-357) how should I use the right owner task in airflow?
[ https://issues.apache.org/jira/browse/AIRFLOW-357?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15389182#comment-15389182 ] wei.he commented on AIRFLOW-357: I figured out this issue. Because I set the AIRFLOW_HOME in /home/airflow/, only airflow can access this file directory. > how should I use the right owner task in airflow? > - > > Key: AIRFLOW-357 > URL: https://issues.apache.org/jira/browse/AIRFLOW-357 > Project: Apache Airflow > Issue Type: Bug >Affects Versions: Airflow 1.7.1 >Reporter: wei.he > > I dont understand the "owner" in airflow. the comment of ower is "the owner > of the task, using the unix username is recommended". I wrote some the > following code. > Default_args = { > 'owner': 'max', > 'depends_on_past': False, > 'start_date': datetime(2016, 7, 14), > 'email': ['m...@test.com'], > 'email_on_failure': False, > 'email_on_retry': False, > 'retries': 1, > 'retry_delay': timedelta(minutes=5), > dag = DAG('dmp-annalect', default_args=default_args, > schedule_interval='30 0 * * *') > task1_pigjob_basedata = """ > {local_dir}/src/basedata/basedata.sh > {local_dir}/log/basedata/run_log & > """.format(local_dir=WORKSPACE) > task1_pigjob_basedata = BashOperator( > task_id='task1_pigjob_basedata_impclk',owner='max', > bash_command=pigjob_basedata_impclk, > dag=dag) > I used the command "airflow test dagid taskid 2016-07-20" , > But I got a error, > ... {bash_operator.py:77} INFO - put: Permission denied: user=airflow, > I thought that my job ran with "max" user, but apperently , ran test using > 'airflow' user . > I hope if I run my task using 'max' user, how should I do. > Thanks -- This message was sent by Atlassian JIRA (v6.3.4#6332)
incubator-airflow git commit: [AIRFLOW-348] Fix code style warnings
Repository: incubator-airflow Updated Branches: refs/heads/master b6e609824 -> 189e6b887 [AIRFLOW-348] Fix code style warnings Closes #1672 from skudriashev/airflow-348 Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/189e6b88 Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/189e6b88 Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/189e6b88 Branch: refs/heads/master Commit: 189e6b88742ace8c46e72d59d7662284e34b7a2e Parents: b6e6098 Author: Stanislav Kudriashev Authored: Fri Jul 22 13:03:45 2016 +0200 Committer: Bolke de Bruin Committed: Fri Jul 22 13:04:03 2016 +0200 -- airflow/configuration.py| 7 ++--- airflow/contrib/hooks/qubole_hook.py| 1 - airflow/example_dags/example_python_operator.py | 11 +++ airflow/hooks/__init__.py | 1 - airflow/hooks/dbapi_hook.py | 15 +- airflow/hooks/hdfs_hook.py | 13 - airflow/hooks/jdbc_hook.py | 8 ++ airflow/hooks/webhdfs_hook.py | 16 +-- airflow/macros/__init__.py | 1 - airflow/models.py | 4 +-- airflow/operators/__init__.py | 1 - airflow/operators/dagrun_operator.py| 1 + airflow/settings.py | 5 +++- airflow/utils/file.py | 12 airflow/utils/helpers.py| 2 ++ airflow/www/views.py| 30 +--- 16 files changed, 58 insertions(+), 70 deletions(-) -- http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/189e6b88/airflow/configuration.py -- diff --git a/airflow/configuration.py b/airflow/configuration.py index e03b713..5a380ae 100644 --- a/airflow/configuration.py +++ b/airflow/configuration.py @@ -661,10 +661,9 @@ def mkdir_p(path): else: raise AirflowConfigException('Had trouble creating a directory') -""" -Setting AIRFLOW_HOME and AIRFLOW_CONFIG from environment variables, using -"~/airflow" and "~/airflow/airflow.cfg" respectively as defaults. -""" + +# Setting AIRFLOW_HOME and AIRFLOW_CONFIG from environment variables, using +# "~/airflow" and "~/airflow/airflow.cfg" respectively as defaults. if 'AIRFLOW_HOME' not in os.environ: AIRFLOW_HOME = expand_env_var('~/airflow') http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/189e6b88/airflow/contrib/hooks/qubole_hook.py -- diff --git a/airflow/contrib/hooks/qubole_hook.py b/airflow/contrib/hooks/qubole_hook.py index 57d00b5..694b12f 100755 --- a/airflow/contrib/hooks/qubole_hook.py +++ b/airflow/contrib/hooks/qubole_hook.py @@ -84,7 +84,6 @@ class QuboleHook(BaseHook): if self.cmd.status != 'done': raise AirflowException('Command Id: {0} failed with Status: {1}'.format(self.cmd.id, self.cmd.status)) - def kill(self, ti): """ Kill (cancel) a Qubole commmand http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/189e6b88/airflow/example_dags/example_python_operator.py -- diff --git a/airflow/example_dags/example_python_operator.py b/airflow/example_dags/example_python_operator.py index 6c0b93f..c5d7193 100644 --- a/airflow/example_dags/example_python_operator.py +++ b/airflow/example_dags/example_python_operator.py @@ -34,7 +34,7 @@ dag = DAG( def my_sleeping_function(random_base): -'''This is a function that will run within the DAG execution''' +"""This is a function that will run within the DAG execution""" time.sleep(random_base) @@ -49,15 +49,12 @@ run_this = PythonOperator( python_callable=print_context, dag=dag) +# Generate 10 sleeping tasks, sleeping from 0 to 9 seconds respectively for i in range(10): -''' -Generating 10 sleeping task, sleeping from 0 to 9 seconds -respectively -''' task = PythonOperator( -task_id='sleep_for_'+str(i), +task_id='sleep_for_' + str(i), python_callable=my_sleeping_function, -op_kwargs={'random_base': float(i)/10}, +op_kwargs={'random_base': float(i) / 10}, dag=dag) task.set_upstream(run_this) http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/189e6b88/airflow/hooks/__init__.py -- diff --git a/airflow/hooks/__init__.py b/airflow/hooks/__init__.py index 4c1891d..883018d 100644 --- a/airflow/hooks/__init__.py +++ b/airflow/hoo
[jira] [Resolved] (AIRFLOW-348) Fix code style warnings
[ https://issues.apache.org/jira/browse/AIRFLOW-348?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bolke de Bruin resolved AIRFLOW-348. Resolution: Fixed > Fix code style warnings > --- > > Key: AIRFLOW-348 > URL: https://issues.apache.org/jira/browse/AIRFLOW-348 > Project: Apache Airflow > Issue Type: Improvement >Affects Versions: Airflow 1.7.1.2 >Reporter: Stanislav Kudriashev >Assignee: Stanislav Kudriashev >Priority: Minor > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (AIRFLOW-348) Fix code style warnings
[ https://issues.apache.org/jira/browse/AIRFLOW-348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15389324#comment-15389324 ] ASF subversion and git services commented on AIRFLOW-348: - Commit 189e6b88742ace8c46e72d59d7662284e34b7a2e in incubator-airflow's branch refs/heads/master from [~skudriashev] [ https://git-wip-us.apache.org/repos/asf?p=incubator-airflow.git;h=189e6b8 ] [AIRFLOW-348] Fix code style warnings Closes #1672 from skudriashev/airflow-348 > Fix code style warnings > --- > > Key: AIRFLOW-348 > URL: https://issues.apache.org/jira/browse/AIRFLOW-348 > Project: Apache Airflow > Issue Type: Improvement >Affects Versions: Airflow 1.7.1.2 >Reporter: Stanislav Kudriashev >Assignee: Stanislav Kudriashev >Priority: Minor > -- This message was sent by Atlassian JIRA (v6.3.4#6332)
incubator-airflow git commit: AIRFLOW-261 Add bcc and cc fields to EmailOperator
Repository: incubator-airflow Updated Branches: refs/heads/master 189e6b887 -> 7628a8656 AIRFLOW-261 Add bcc and cc fields to EmailOperator Closes #1670 from ajayyadava/261 Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/7628a865 Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/7628a865 Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/7628a865 Branch: refs/heads/master Commit: 7628a8656c37110ef9ea33c5142b52305ebe9440 Parents: 189e6b8 Author: Ajay Yadava Authored: Fri Jul 22 13:08:52 2016 +0200 Committer: Bolke de Bruin Committed: Fri Jul 22 13:08:56 2016 +0200 -- airflow/operators/email_operator.py | 10 +++- airflow/utils/email.py | 39 +++- tests/core.py | 20 +++- 3 files changed, 56 insertions(+), 13 deletions(-) -- http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/7628a865/airflow/operators/email_operator.py -- diff --git a/airflow/operators/email_operator.py b/airflow/operators/email_operator.py index 91a8d05..76cc56b 100644 --- a/airflow/operators/email_operator.py +++ b/airflow/operators/email_operator.py @@ -30,6 +30,10 @@ class EmailOperator(BaseOperator): :type html_content: string :param files: file names to attach in email :type files: list +:param cc: list of recipients to be added in CC field +:type cc: list or string (comma or semicolon delimited) +:param bcc: list of recipients to be added in BCC field +:type bcc: list or string (comma or semicolon delimited) """ template_fields = ('subject', 'html_content') @@ -43,12 +47,16 @@ class EmailOperator(BaseOperator): subject, html_content, files=None, +cc=None, +bcc=None, *args, **kwargs): super(EmailOperator, self).__init__(*args, **kwargs) self.to = to self.subject = subject self.html_content = html_content self.files = files or [] +self.cc = cc +self.bcc = bcc def execute(self, context): -send_email(self.to, self.subject, self.html_content, files=self.files) +send_email(self.to, self.subject, self.html_content, files=self.files, cc=self.cc, bcc=self.bcc) http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/7628a865/airflow/utils/email.py -- diff --git a/airflow/utils/email.py b/airflow/utils/email.py index c19bb89..6fe8662 100644 --- a/airflow/utils/email.py +++ b/airflow/utils/email.py @@ -33,17 +33,17 @@ from email.utils import formatdate from airflow import configuration -def send_email(to, subject, html_content, files=None, dryrun=False): +def send_email(to, subject, html_content, files=None, dryrun=False, cc=None, bcc=None): """ Send email using backend specified in EMAIL_BACKEND. """ path, attr = configuration.get('email', 'EMAIL_BACKEND').rsplit('.', 1) module = importlib.import_module(path) backend = getattr(module, attr) -return backend(to, subject, html_content, files=files, dryrun=dryrun) +return backend(to, subject, html_content, files=files, dryrun=dryrun, cc=cc, bcc=bcc) -def send_email_smtp(to, subject, html_content, files=None, dryrun=False): +def send_email_smtp(to, subject, html_content, files=None, dryrun=False, cc=None, bcc=None): """ Send an email with html content @@ -51,18 +51,23 @@ def send_email_smtp(to, subject, html_content, files=None, dryrun=False): """ SMTP_MAIL_FROM = configuration.get('smtp', 'SMTP_MAIL_FROM') -if isinstance(to, basestring): -if ',' in to: -to = to.split(',') -elif ';' in to: -to = to.split(';') -else: -to = [to] +to = get_email_address_list(to) msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = SMTP_MAIL_FROM msg['To'] = ", ".join(to) +recipients = to +if cc: +cc = get_email_address_list(cc) +msg['CC'] = ", ".join(cc) +recipients = recipients + cc + +if bcc: +# don't add bcc in header +bcc = get_email_address_list(bcc) +recipients = recipients + bcc + msg['Date'] = formatdate(localtime=True) mime_text = MIMEText(html_content, 'html') msg.attach(mime_text) @@ -76,7 +81,7 @@ def send_email_smtp(to, subject, html_content, files=None, dryrun=False): Name=basename )) -send_MIME_email(SMTP_MAIL_FROM, to, msg, dryrun) +send_MIME_email(SMTP_MAIL_FROM, recipients, msg, dryrun) def send_M
[jira] [Commented] (AIRFLOW-261) Add more fields for EmailOperator
[ https://issues.apache.org/jira/browse/AIRFLOW-261?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15389330#comment-15389330 ] ASF subversion and git services commented on AIRFLOW-261: - Commit 7628a8656c37110ef9ea33c5142b52305ebe9440 in incubator-airflow's branch refs/heads/master from [~ajayyadava] [ https://git-wip-us.apache.org/repos/asf?p=incubator-airflow.git;h=7628a86 ] AIRFLOW-261 Add bcc and cc fields to EmailOperator Closes #1670 from ajayyadava/261 > Add more fields for EmailOperator > - > > Key: AIRFLOW-261 > URL: https://issues.apache.org/jira/browse/AIRFLOW-261 > Project: Apache Airflow > Issue Type: Improvement >Reporter: Ajay Yadava >Assignee: Ajay Yadava > > It will be nice to be able to specify cc, bcc recipients for the > EmailOperator. -- This message was sent by Atlassian JIRA (v6.3.4#6332)