Thanks Bolke. For those that are not aware, my team is working with Bolke's team on Amundsen which is a data discovery and metadata project( https://github.com/lyft/amundsen) . I think although it ships with Atlas client(or it used to be), the new API per my understanding is generic enough that doesn't tight with atlas. E.g we(Lyft) could build a neo4j / Amundsen client in our Airflow fork to ingest the lineage info in a push fashion to build the lineage.
Amundsen itself has put up the effort to integrate Airflow with the tool(connect which DAG/task produces the data set etc). With this change, I foresee it will help to provide more enriched metadata. Thanks, -Tao On Wed, Jan 22, 2020 at 8:46 AM Dan Davydov <ddavy...@twitter.com.invalid> wrote: > Just want to preface my reply with the fact that I haven't thought about > data lineage very much. > > This is an awesome idea :)! I like something like 1) personally, e.g. > operators could optionally define a .outlet() and .inlet() interface which > would return the inlets and outlets of a given task, and then it's up to > the operator how it wants to set these inlets/outlets like the Papermill > operator currently does. This also keeps allows inlets/outlets more dynamic > (e.g. in the case of an operator that might generate inlets/outlets > dynamically at execution time). Seems the most extensible/least coupling. > IMO we should strive to make DAGs easy to create with little boilerplate, > but this is a lot less important for operators since they are a lot more > stable and change less frequently, so it's fine to require operators to > implement some interface manually. > > On Wed, Jan 22, 2020 at 8:33 AM Bolke de Bruin <bdbr...@gmail.com> wrote: > > > Dear All, > > > > Over last few weeks I made serious improvements to the lineage support > that > > Airflow has. Whilst not complete it’s starting to shape up and I think it > > is good to share some thoughts and directions. Much has been discussed > with > > several organisations like Polidea, Daily Motion and Lyft. Some have > > already implemented some support for lineage themselves (Daily Motion) > and > > some have a need for it (Lyft with Amundsen). > > > > First a bit of a recap. What is lineage of why is it important? Lineage > > allows you to track the origins of data what happens to it and where it > > moves over time. Lineage is often associated with audibility of data > > pipelines which is not a very sexy subject ;-). However, there are much > > more prominent and user facing improvements possible if you have lineage > > data available. Lineage greatly simplifies the ability to trace back > errors > > to the root cause in analytics. So, instead of the user calling up the > > engineering team in case of a data error, it could traceback to the > origin > > of the data and call the one that has created the original data set. > > Lineage also greatly improves discoverability of data. Lineage > information > > gives insights into the importance of data sets. So if a new employee > joins > > a team he would normally go to the most senior person in that team to ask > > him what data sources he is using and what their meaning is. If lineage > > information is exposed through a tool like Amundsen this is not required > > because that person can just look it up. > > > > To summarise their are 3 use cases driving the need for lineage: > > > > 1. Discoverability of data > > 2. Improved data operations > > 3. Audibility of data pipelines > > > > So that’s all great I hear you thinking, but why don’t we have it in > > Airflow already if it is so important? The answer to that is two fold. > > Firstly, adding lineage information is often associated with a lot of > > metadata and meta programming. Typically if lineage is being ’slapped on’ > > one needs to add a lot of metadata which then need to be kept in sync. In > > that way it does not solve a problem for the developer and rather it > > creates one. Secondly, Airflow is a task based system and by definition > > does not have a very good infrastructure that deals with data. In the > past > > we had some trials by Jeremiah to add Pipelines, but it never was > > integrated and I think it actually sparked him to start Prefect ;-) > > (correct me if I am wrong if you are reading this Jermiah). > > > > Where is lineage support now in Airflow? In the 1.10.X series there is > some > > support for lineage, but it is buggy and difficult to use as it is based > on > > the metadata model of Apache Atlas. In master the foundation has much > > improved (but fully done yet). You can now set inlets and outlets with > > lightweight objects like File(url=“http://www.google.com”) and > > Table(name=“my_table”) and the lineage system in Airflow will figure out > a > > lot for you. You can also have inlets pick up outlets from previous > > upstream tasks by passing a list of task_ids or even using “AUTO” which > > picks up outlets from direct upstream tasks. > > > > The lightweight objects are automatically templated so you can do > something > > like File(url=“/tmp/my_data_{{ execution_date }}”) which does the right > > thing for you. Templating inlets and outlets gives very powerful > > capabilities by for example creating a Task, that, based on the inlets it > > receives, can drop PII information from an arbitrary table and output > this > > table somewhere else. This allows for creating Generic Tasks/Dags that > can > > be re-used without any domain knowledge. A small example (not PII) is > > available with the example_papermill_operator. > > > > Lineage information is exposed through an API endpoint. You can query > > “/api/experimental/lineage/<dag_id>/<execution_date>” and you will get a > > list of tasks with their inlets and outlets defined. The lineage > > information shared through the API and the lightweight object model are > > very close to the model used within Lyft’s Amundsen so when that gets > > proper visualisation support for lineage and pulls in the information > from > > Airflow it’s presto! Other systems might require some translation but > that > > shouldn’t be too hard. > > > > What doesn’t it do? Well, and here we get to the point of this > discussion, > > there is still meta programming involved to keep the normal parameters > and > > the inlets and outlets to an operator in sync. This is because it’s hard > to > > make operators lineage aware without changing them. So while you set > > “inlets” and “outlets” to an Operator the operator itself doesn’t do > > anything with them, making them a lot less powerful. Actually, there is > > only one operator that has out of the box support for lineage is the > > PapermillOperator. > > > > In discussions with the aforementioned organisations it became clear > that, > > while we could change all operators that Airflow comes out of the box > with, > > this will not help with the many custom operators that are around. They > > will simply not get updated as part of this exercise, leaving them as > > technical debt. Thus we need an approach that works with the past and > > improves the future. The generic pattern for Airflow operators is pretty > > simple: you can read many (yes we know there are exceptions!) as > > SourceToTarget(src_conn_id, src_xxx, src_xx, target_conn_id, target_xxx, > > some_other_kwarg). Hence, we came up with the following: > > > > For existing non lineage aware operators: > > > > 1. Use wrapper objects to group parameters together as inlet or as > outlet. > > For example usage for the MysqlToHiveTransfer could look like > > MysqlToHiveTransfer(Inlet(mysql_conn_id=‘mysql_conn’, sql=’select * from > > table’), Outlet(hive_cli_conn_id=‘hive_conn’, > hive_table=‘my_hive_table’)). > > The wrapper objects would then set the right kwargs to the Operator and > > create the lineage information. This resolves the issue of keeping > > parameters in sync. > > 2. Use the build pattern to tell the lineage system which arguments to > the > > operator are for the Inlet and for the Outlet. Maybe with a type hint if > > required. E.g. > > MysqlToHiveTransfer(mysql_conn_id=‘conn_id’, sql=’select * from table’, > > hive_cli_conn_id=‘hive_conn’, > > hive_table=‘hive_table’).inlet(‘mysql_conn_id’,{’sql’: > > ‘mysql’}).outlet(‘hive_cli_conn_id’, ‘hive_table’) > > This requires a bit more work from the developer as the parameter names > > need to be kept in sync. However, they are slow moving. > > > > Future lineage aware operators: > > > > 1. Update the Operator to set and support inlets and outlets itself. E.g. > > like the current PapermillOperator > > 2. Have a dictionary inside the operator which tells the lineage system > > what fields are used for inlet and outlet. This is the integrated pattern > > of 2 for non lineage aware operators: > > # dictionary of parameter name with type > > inlet_fields = {‘mysql_conn_id’: ‘mysql_connection’, ’sql’: ’sql’} > > outlet_fields = {‘hive_conn_id’: ‘hive_connection’, ’hive_table’’: > > ’table’} > > Updates to the operator need to be checked to ensure the fields names are > > kept in sync. > > 3. Enforce a naming pattern for Operators like > > MysqlToHiveTransfer(…) becomes > > MysqlToHive(mysql_conn_id, mysql_sql, hive_conn_id, hive_table) or > > MysqlToHive(src_conn_id, src_sql, target_conn_id, target_table) > > This would allow the lineage system to figure out what is inlet and what > is > > outlet based on the naming scheme. It would require pylint plugin to make > > sure Operators to behave correctly, but would also make operators much > more > > predictable. > > > > Option number 3 for the future has the most impact. Out of the box the > > lineage system in Airflow can support (and its my intention to do so) all > > the above patterns, but ideally we do improve the state so that we can > > deprecate what we do for non lineage aware operators in the future: > wrapper > > objects and the build pattern wouldn’t be necessary anymore. > > > > What do you think? What are your thoughts on lineage, what kind of usages > > do you foresee? How would you like to be using it and have it supported > in > > Airflow? Would you be able to work with the above ways of doing it? Pros > > and cons? > > > > Thanks > > Bolke > > >
