[AIRFLOW-2463] Make task instance context available for hive queries by KevinYang21 · Pull Request #3405 · apache/airflow

KevinYang21 · 2018-05-23T01:16:31Z

Make sure you have checked all steps below.

JIRA

My PR addresses the following Airflow JIRA issues and references them in the PR title.

Description

Here are some details about my PR, including screenshots of any UI changes:

This PR is mainly to enable hive (e.g. through hive audit hook) to log task instance context( dag_id, task_id, dag_run_id and task instance execution_date) for hive queries. This would give users more audit and research power. For example:

When data in one table was polluted, people can find out what tasks using the table need to rerun.
Grouping hive queries by dag_id and task_id.
Figure out whether the DAG run of the task instance is a manual run, scheduled run or backfill run.

This change will make task instance context available in the following cases:

HiveServer2Hook() calls in Airflow official operators: as hive_conf
HiveCliHook() and HiveServer2Hook() calls in PythonOperator: as hive_conf
hive cli calls in BashOperator: as environment variable

The following cases are already covered without this change (with updated names):

HiveCliHook() calls in HiveOperator(): as hive_conf

The following cases are not cover even with this change:

beeline calls in BashOperator.

NOTE:
If as environment variable, context would be available as AIRFLOW_CTX_DAG_ID, AIRFLOW_CTX_TASK_ID, AIRFLOW_CTX_EXECUTION_DATE, AIRFLOW_CTX_DAG_RUN_ID.
If as hvie_conf, as airflow.ctx.dag_id, airflow.ctx.task_id, airflow.ctx.execution_date, airflow.ctx.dag_run_id. This is to follow the environment variable and hive conf naming convention.

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
-- tests/operators/bash_operator.py
-- tests/operators/python_operator.py:PythonOperatorTest.test_echo_env_variables
-- tests/hooks/test_hive_hook.py: TestHiveCliHook.test_run_cli_with_hive_conf
-- tests/hooks/test_hive_hook.py: TestHiveServer2Hook. test_get_results_with_hive_conf

HiveServer2Hook uses impyla, which is compatible with at most thrift==0.9.3, but we are using thrift==0.11.0 in our CI. However we cannot just downgrade thrift because our new client package used in HiveCliHook, hmsclient, cannot work with thrift<=0.9.3. Thus the unit tests for HiveSever2Hook are removed and tested manually instead.

This logic has been running in Airbnb production for a week.

Manually tested with test DAG file: https://gist.github.com/yrqls21/a89e37b82b5fc58a2e189d8ee23e61e4

airflow test output:
hive_cli task:

[2018-05-17 02:22:19,103] {hive_hooks.py:243} INFO - airflow.ctx.dag_id=kevin_hive_s2_test
[2018-05-17 02:22:19,105] {hive_hooks.py:243} INFO - airflow.ctx.dag_run_id=
[2018-05-17 02:22:19,106] {hive_hooks.py:243} INFO - airflow.ctx.task_id=hive_cli
[2018-05-17 02:22:19,106] {hive_hooks.py:243} INFO - airflow.ctx.execution_date=2018-05-10T00:00:00

beeline task:

[2018-05-17 02:22:50,442] {hive_hooks.py:243} INFO - 0: jdbc:hive2://i-0e2d3cb2c448a2300:10000/def> set airflow.ctx.dag_id;
[2018-05-17 02:22:50,512] {hive_hooks.py:243} INFO - +----------------------------------------+--+
[2018-05-17 02:22:50,512] {hive_hooks.py:243} INFO - |                  set                   |
[2018-05-17 02:22:50,513] {hive_hooks.py:243} INFO - +----------------------------------------+--+
[2018-05-17 02:22:50,513] {hive_hooks.py:243} INFO - | airflow.ctx.dag_id=kevin_hive_s2_test  |
[2018-05-17 02:22:50,513] {hive_hooks.py:243} INFO - +----------------------------------------+--+
[2018-05-17 02:22:50,513] {hive_hooks.py:243} INFO - 1 row selected (0.07 seconds)
[2018-05-17 02:22:50,514] {hive_hooks.py:243} INFO - 0: jdbc:hive2://i-0e2d3cb2c448a2300:10000/def> set airflow.ctx.dag_run_id;
[2018-05-17 02:22:50,516] {hive_hooks.py:243} INFO - +--------------------------+--+
[2018-05-17 02:22:50,516] {hive_hooks.py:243} INFO - |           set            |
[2018-05-17 02:22:50,517] {hive_hooks.py:243} INFO - +--------------------------+--+
[2018-05-17 02:22:50,517] {hive_hooks.py:243} INFO - | airflow.ctx.dag_run_id=  |
[2018-05-17 02:22:50,517] {hive_hooks.py:243} INFO - +--------------------------+--+
[2018-05-17 02:22:50,517] {hive_hooks.py:243} INFO - 1 row selected (0.002 seconds)
[2018-05-17 02:22:50,517] {hive_hooks.py:243} INFO - 0: jdbc:hive2://i-0e2d3cb2c448a2300:10000/def> set airflow.ctx.task_id;
[2018-05-17 02:22:50,520] {hive_hooks.py:243} INFO - +------------------------------+--+
[2018-05-17 02:22:50,520] {hive_hooks.py:243} INFO - |             set              |
[2018-05-17 02:22:50,521] {hive_hooks.py:243} INFO - +------------------------------+--+
[2018-05-17 02:22:50,521] {hive_hooks.py:243} INFO - | airflow.ctx.task_id=beeline  |
[2018-05-17 02:22:50,521] {hive_hooks.py:243} INFO - +------------------------------+--+
[2018-05-17 02:22:50,526] {hive_hooks.py:243} INFO - 1 row selected (0.009 seconds)
[2018-05-17 02:22:50,527] {hive_hooks.py:243} INFO - 0: jdbc:hive2://i-0e2d3cb2c448a2300:10000/def> set airflow.ctx.execution_date;
[2018-05-17 02:22:50,530] {hive_hooks.py:243} INFO - +-------------------------------------------------+--+
[2018-05-17 02:22:50,530] {hive_hooks.py:243} INFO - |                       set                       |
[2018-05-17 02:22:50,530] {hive_hooks.py:243} INFO - +-------------------------------------------------+--+
[2018-05-17 02:22:50,531] {hive_hooks.py:243} INFO - | airflow.ctx.execution_date=2018-05-10T00:00:00  |
[2018-05-17 02:22:50,531] {hive_hooks.py:243} INFO - +-------------------------------------------------+--+
[2018-05-17 02:22:50,531] {hive_hooks.py:243} INFO - 1 row selected (0.003 seconds)

hive_in_python_op task:

[2018-05-17 02:23:26,762] {python_operator.py:92} INFO - Exporting the following env vars:                                                                                     AIRFLOW_CTX_TASK_ID=hive_in_python_op
AIRFLOW_CTX_DAG_ID=kevin_hive_s2_test
AIRFLOW_CTX_EXECUTION_DATE=2018-05-10T00:00:00
[2018-05-15 08:17:28,011] {hive_hooks.py:246} INFO - AIRFLOW_CTX_DAG_ID=kevin_hive_s2_test
[2018-05-15 08:17:32,560] {hive_hooks.py:246} INFO - AIRFLOW_CTX_DAG_RUN_ID=
[2018-05-15 08:17:37,111] {hive_hooks.py:246} INFO - AIRFLOW_CTX_TASK_ID=hive_in_python_opp
[2018-05-15 08:17:41,725] {hive_hooks.py:246} INFO - AIRFLOW_CTX_EXECUTION_DATE=2018-05-10T00:00:00
OK
Time taken: 0.917 seconds
airflow.ctx.dag_id=kevin_hive_s2_test
OK
Time taken: 0.892 seconds
airflow.ctx.dag_run_id=
OK
Time taken: 0.869 seconds
airflow.ctx.task_id=hive_in_python_op
OK
Time taken: 0.895 seconds
airflow.ctx.execution_date=2018-05-10T00:00:00

echo_env_vars task:

[2018-05-17 02:25:53,249] {bash_operator.py:79} INFO - Exporting the following env vars:
AIRFLOW_CTX_TASK_ID=echo_env_vars
AIRFLOW_CTX_DAG_ID=kevin_hive_s2_test
AIRFLOW_CTX_EXECUTION_DATE=2018-05-10T00:00:00
[2018-05-17 02:25:53,259] {bash_operator.py:101} INFO - Output:
[2018-05-17 02:25:53,261] {bash_operator.py:105} INFO - kevin_hive_s2_test
[2018-05-17 02:25:53,261] {bash_operator.py:105} INFO - 2018-05-10T00:00:00
[2018-05-17 02:25:53,261] {bash_operator.py:105} INFO - echo_env_vars
[2018-05-17 02:25:53,261] {bash_operator.py:105} INFO -
[2018-05-17 02:25:53,261] {bash_operator.py:105} INFO - /home/kevin_yang/airflow
[2018-05-17 02:25:53,262] {bash_operator.py:108} INFO - Command exited with return code 0

Commits

My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"

Documentation

In case of new functionality, my PR adds documentation that describes how to use it.
- When adding new operators/hooks/sensors, the autoclass documentation generation needs to be added.

Code Quality

Passes git diff upstream/master -u -- "*.py" | flake8 --diff

codecov-io · 2018-05-24T01:31:32Z

Codecov Report

Merging #3405 into master will increase coverage by 0.01%.
The diff coverage is 93.22%.

@@            Coverage Diff             @@
##           master    #3405      +/-   ##
==========================================
+ Coverage   77.45%   77.46%   +0.01%     
==========================================
  Files         204      204              
  Lines       15299    15326      +27     
==========================================
+ Hits        11850    11873      +23     
- Misses       3449     3453       +4

Impacted Files	Coverage Δ
airflow/operators/hive_to_samba_operator.py	`0% <0%> (ø)`	⬆️
airflow/operators/hive_to_mysql.py	`0% <0%> (ø)`	⬆️
airflow/operators/python_operator.py	`95.03% <100%> (+0.12%)`	⬆️
airflow/hooks/hive_hooks.py	`75.26% <100%> (+0.6%)`	⬆️
airflow/utils/operator_helpers.py	`100% <100%> (ø)`	⬆️
airflow/operators/bash_operator.py	`91.37% <100%> (+0.99%)`	⬆️
airflow/utils/state.py	`93.33% <0%> (-3.34%)`	⬇️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update b0061f1...35345fd. Read the comment docs.

Fokko · 2018-05-24T08:36:04Z

Could you rebase against master? I see a lot of changes, also some files that look unrelated, like the python operator.

@gglanzani Hive master, can you take a peek?

KevinYang21 · 2018-05-24T08:57:19Z

Thank you @Fokko for taking a look. I was working on making the PR pass the CI, will update the PR description later to provide more context on this change.

KevinYang21 · 2018-05-25T00:47:49Z

PTAL @saguziel @Fokko @afernandez

KevinYang21 · 2018-05-24T19:17:20Z

Needed to make beeline hook work.

Since this is just a string, can you write it as

jdbc_url = '"{}"'.format(jdbc_url)

as to avoid escaping?

sure thing, will update

KevinYang21 · 2018-05-24T19:17:45Z

Needed to make beeline hook work properly (otherwise last statement won't be executed).

gglanzani

WIP review

gglanzani · 2018-05-26T13:36:11Z

Call me a moron, but I prefer dict comprehension here

KevinYang21 · 2018-06-09T20:00:07Z

@Fokko Sure I agree the situation in bash operator is a bit independent and I'm fine opening another PR just for the change of behavior on how do we get AIRFLOW_HOME value--I can elaborate more on why I chose to do that in the PR. But for the python operator part, I feel opening another PR making changes to provide Airflow context in hive is missing some context and thus belongs to this PR more--updating the env var mapping is quite safe and I don't this worth losing the context to create another PR for it.

Also, would you elaborate a bit more on what checks do you have in mind please?

Fokko · 2018-06-09T20:11:11Z

The most important thing is that we don't change current behaviour, we can extend it, but it should be backwards compatible.

KevinYang21 · 2018-06-10T08:11:58Z

@Fokko I moved the behavior changes in bash operator out of this PR and create this PR just for that. I listed the reason why I think we should make this change there in the PR description, let's move the discussion on that there.

KevinYang21

Rebased to pyhive, also added unit test for hive server2 hook since the blocking dependency issue was lifted by switching to pyhive.

@Fokko @gglanzani @saguziel PTAL

KevinYang21 · 2018-06-28T09:18:25Z

@gglanzani adding the condition here for queries like set key

@yrqls21 Pardon my ignorance – I couldn't get Google to help me either – what kind of queries do start with set key?

I meant those query to get variable value like 'set airflow.ctx.task_id' would return the value set by set airflow.ctx.task_id=some_id.

KevinYang21 · 2018-06-28T09:23:03Z

@gglanzani had to extract the actually result string from a result tuple after rebased to pyhive, while we use to get the result string directly when using impyla. Should we be concerned about this change or make it to UPDATING.md( or maybe u have done that and I missed)?

@yrqls21 I understand what you mean. PyHive returns indeed always tuples, and I call list on them, i.e. we always get structures as

[(1,), (2,)]

I did not know that Impyla returned a list of elements in the above case. I don't have a working impyla installation to test that.

The DB 2 API (that Impyla should have been following and PyHive is following) prescribes that the results should be a sequence of sequences. We are doing the proper thing here IMHO.

The fact remains that this behaviour might be unexpected when updating, so I guess we need to update UPDATING.md (I haven't done it).

Is it possible to do it here? I think this text would suffice:

HiveServer2Hook.get_results() always returns a list of tuples, even when a single column is queried, as per Python API 2.

@Fokko: Agree?

I agree if that's what DB2 API specifies.

KevinYang21 · 2018-07-03T23:43:26Z

ping

KevinYang21 · 2018-07-11T02:34:13Z

ping

gglanzani · 2018-07-11T05:18:57Z

@Fokko lgtm

Fokko

LGTM

[AIRFLOW-2463] Make task instance context available for hive queries update UPDATING.md, please squash Closes apache#3405 from yrqls21/kevin_yang_add_context

KevinYang21 force-pushed the kevin_yang_add_context branch 8 times, most recently from e830fb5 to 9815017 Compare May 24, 2018 00:53

KevinYang21 force-pushed the kevin_yang_add_context branch 9 times, most recently from 7ff263e to fc7467f Compare May 24, 2018 07:24

KevinYang21 force-pushed the kevin_yang_add_context branch from fc7467f to 266a27a Compare May 24, 2018 08:56

KevinYang21 force-pushed the kevin_yang_add_context branch from 266a27a to b8237aa Compare May 24, 2018 09:23

KevinYang21 changed the title ~~[WIP][AIRFLOW-2463] Make task instance context available for hive queries~~ [AIRFLOW-2463] Make task instance context available for hive queries May 24, 2018

KevinYang21 commented May 25, 2018

View reviewed changes

KevinYang21 force-pushed the kevin_yang_add_context branch from b8237aa to 3b7ca79 Compare May 26, 2018 01:25

gglanzani reviewed May 26, 2018

View reviewed changes

Comment thread airflow/hooks/hive_hooks.py Outdated

gglanzani May 26, 2018

Copy link
Copy Markdown

Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Call me a moron, but I prefer dict comprehension here

KevinYang21 force-pushed the kevin_yang_add_context branch 3 times, most recently from 8476d26 to 8d0f72f Compare May 29, 2018 22:02

KevinYang21 force-pushed the kevin_yang_add_context branch 2 times, most recently from b9222c8 to 1eeba1b Compare June 10, 2018 04:13

KevinYang21 changed the title ~~[AIRFLOW-2463] Make task instance context available for hive queries~~ [WIP][AIRFLOW-2463] Make task instance context available for hive queries Jun 19, 2018

KevinYang21 force-pushed the kevin_yang_add_context branch 11 times, most recently from 94d1c56 to be2ad65 Compare June 28, 2018 09:17

KevinYang21 commented Jun 28, 2018

View reviewed changes

KevinYang21 changed the title ~~[WIP][AIRFLOW-2463] Make task instance context available for hive queries~~ [AIRFLOW-2463] Make task instance context available for hive queries Jun 28, 2018

[AIRFLOW-2463] Make task instance context available for hive queries

35345fd

KevinYang21 force-pushed the kevin_yang_add_context branch from be2ad65 to 35345fd Compare June 28, 2018 10:11

update UPDATING.md, please squash

57fab01

Fokko approved these changes Jul 11, 2018

View reviewed changes

asfgit closed this in e9babff Jul 11, 2018

r-richmond mentioned this pull request Apr 30, 2019

[AIRFLOW-4434] fix support for impala in hive hook #5206

Merged

6 tasks done

feng-tao mentioned this pull request Oct 22, 2019

[AIRFLOW-5715] Make email, owner context available #6385

Merged

2 of 5 tasks

Conversation

KevinYang21 commented May 23, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

JIRA

Description

Tests

Commits

Documentation

Code Quality

Uh oh!

codecov-io commented May 24, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Fokko commented May 24, 2018

Uh oh!

KevinYang21 commented May 24, 2018

Uh oh!

KevinYang21 commented May 25, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gglanzani left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

KevinYang21 commented Jun 9, 2018

Uh oh!

Fokko commented Jun 9, 2018

Uh oh!

KevinYang21 commented Jun 10, 2018

Uh oh!

KevinYang21 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gglanzani Jun 28, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

KevinYang21 commented Jul 3, 2018

Uh oh!

KevinYang21 commented Jul 11, 2018

Uh oh!

gglanzani commented Jul 11, 2018

Uh oh!

Fokko left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

KevinYang21 commented May 23, 2018 •

edited

Loading

codecov-io commented May 24, 2018 •

edited

Loading

gglanzani Jun 28, 2018 •

edited

Loading