<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Datapackage Pipelines Tutorial
This tutorial is built as a Jupyter notebook which allows you to run and modify the code inline and can be used as a starting point for new Datapackage Pipelines projects.
Installation
Follow the DataFlows Tutorial installation instructions.
Save this tutorial in curreny working directory (right-click and save on following link)
Step1:  This installs datapackage-pipelines with speed optimizations, if you encounter problems installing it, remove the [speedup] suffix.
Verify you have the latest datapackage-pipelines version
Step2:  Create a flow
Datapackage-pipelines uses the DataFlows library's Flow objects as the basic building blocks for larger pipeline systems.
It's recommended to follow the DataFlows Tutorial to get a better understanding of the DataFlows concepts which will be used here.
Run the following cell to create a file called countries_population_flow.py which scrapes a list of countries populations from wikipedia.
This flow is copied from the DataFlows tutorial, the processing function country_population is exactly the same, the flow and how we run it is changed to integrate with Datapackage Pipelines
Step3:  Run the flow
Step4:  This is standard DataFlows library usage, now let's see what datapackage-pipelines provides
Create a pipeline spec
Datapackage-pipelines uses yaml files to to define pipelines of flow steps.
Create a spec to run the countries population flow and save to a path
Step5:  Using dpp
dpp is the CLI interface to the datapackage-pipelines library. It is used to list and run available pipelines.
Let's list the available pipelines to see if our countries-population pipeline is available
Step6:  Run the pipeline
Step7:  Pipeline Dependencies
Let's add another pipeline which depends on the countries-population pipeline.
This time we will use just the pipeline spec yaml to write the pipeline, without any DataFlows code (although DataFlows library is used to implement the processors we are using here)
Step8:  Clear the pipelines state using dpp init and list the available pipelines
Step9:  You can see that the new pipeline can't run until it's dependency is executed.
Let's run all the "Dirty" dependencies
Step10:  Inspect the created datapackage
Step11:  Pipeline processors
Datapackage Pipelines has a standard library of processors, like the sort processor used previously. These processors correspond to DataFlows standard library processors.
See the Datapackage Pipelines README for reference and usage examples.
An example showing usage of common processors
Step12:  Run the pipeline
Step13:  Print the datapackage
Step14:  Pipelines Server
Running pipelines on your laptop is fine for many use-cases but sometimes you want to run pipelines in a more reproducible, scalable and automatic fashion.
The Datapackage Pipelines Server is a Docker image which provides the core functionality to achieve this.
To start a local pipelines server for development, you will need to install Docker for Windows,
Mac or Linux
Pull the datapackage-pipelines image
Step15:  Start a local pipelines server, mounting the current working directory into the container
Step16:  After a few seconds, the pipelines dashboad should be available at http
Step17:  Inspect the Pipelines server logs and wait for Update Pipelines task to complete and pipelines to start running
<ASSISTANT_TASK:>
Python Code:
%%sh
python3 -m pip install -qU datapackage-pipelines[seedup]
Explanation: Datapackage Pipelines Tutorial
This tutorial is built as a Jupyter notebook which allows you to run and modify the code inline and can be used as a starting point for new Datapackage Pipelines projects.
Installation
Follow the DataFlows Tutorial installation instructions.
Save this tutorial in curreny working directory (right-click and save on following link): https://raw.githubusercontent.com/frictionlessdata/datapackage-pipelines/master/TUTORIAL.ipynb
Start Jupyter Lab in the dataflows environment and open the datapackage pipelines tutorial notebook you downloaded.
Install datapackage-pipelines
End of explanation
%%sh
dpp version
Explanation: This installs datapackage-pipelines with speed optimizations, if you encounter problems installing it, remove the [speedup] suffix.
Verify you have the latest datapackage-pipelines version
End of explanation
%%writefile countries_population_flow.py
# notice that we don't import any datapackage-pipelines modules
# all the flow code is written purely with the DataFlows library
from dataflows import Flow, dump_to_path, load, add_metadata, printer, update_resource
from lxml import etree
from urllib.request import urlopen
# Generator flow step, copied from the DataFlows tutorial
# it just spews rows of data - in this case, countries populations scraped from Wikipedia
def country_population():
    # Read the Wikipedia page and parse it using etree
    page = urlopen('https://en.wikipedia.org/w/index.php?title=List_of_countries_and_dependencies_by_population&oldid=987469839').read()
    parser = etree.XMLParser(recover=True)
    tree = etree.fromstring(page, parser)
    # Iterate on all tables, rows and cells
    for table in tree.findall('.//table'):
        if 'wikitable' in table.attrib.get('class', ''):
            for row in table.find('tbody').findall('tr'):
                cells = row.findall('td')
                if len(cells) > 3:
                    # If a matching row is found...
                    name = cells[0].find('.//a').attrib.get('title').replace("Demographics of","")
                    population = cells[1].text
                    # ... yield a row with the information
                    yield dict(
                        name=name,
                        population=population
                    )
# The main entrypoint for Datapackage Pipelines, each flow file should have a single flow function
def flow(*args):
    return Flow(
        country_population(),
        update_resource('res_1', **{
            # Set a proper name for the resource
            'name': 'countries_population',
            # Always set a path as well, even if you don't intend to save it to the filesystem
            'path': 'countries_population.csv',
            # dpp:streaming property is required to let Datapackage Pipelines know it should handle this resource
            'dpp:streaming': True,
        })
    )
# Entrypoint for running the flow directly, without Datapackage Pipelines
if __name__ == '__main__':
    # Add a printer step and run the flow
    Flow(flow(), printer(num_rows=1, tablefmt='html')).process()
Explanation: Create a flow
Datapackage-pipelines uses the DataFlows library's Flow objects as the basic building blocks for larger pipeline systems.
It's recommended to follow the DataFlows Tutorial to get a better understanding of the DataFlows concepts which will be used here.
Run the following cell to create a file called countries_population_flow.py which scrapes a list of countries populations from wikipedia.
This flow is copied from the DataFlows tutorial, the processing function country_population is exactly the same, the flow and how we run it is changed to integrate with Datapackage Pipelines:
End of explanation
%run countries_population_flow.py
Explanation: Run the flow:
End of explanation
%%writefile pipeline-spec.yaml
countries-population:
  pipeline:
  - flow: countries_population_flow
  - run: dump.to_path
    parameters:
      out-path: data/countries_population
Explanation: This is standard DataFlows library usage, now let's see what datapackage-pipelines provides
Create a pipeline spec
Datapackage-pipelines uses yaml files to to define pipelines of flow steps.
Create a spec to run the countries population flow and save to a path:
End of explanation
%%sh
dpp
Explanation: Using dpp
dpp is the CLI interface to the datapackage-pipelines library. It is used to list and run available pipelines.
Let's list the available pipelines to see if our countries-population pipeline is available:
End of explanation
%%sh
dpp run ./countries-population
Explanation: Run the pipeline:
End of explanation
%%writefile pipeline-spec.yaml
countries-population:
  pipeline:
  - flow: countries_population_flow
  - run: dump.to_path
    parameters:
      out-path: data/countries_population
sorted_countries_by_name:
  dependencies:
  - pipeline: ./countries-population
  - datapackage: data/countries_population/datapackage.json
  pipeline:
  - run: load
    parameters:
      from: data/countries_population/datapackage.json
      resources: ['countries_population']
  - run: sort
    parameters:
      resources: ['countries_population']
      sort-by: '{name}'
  - run: dump.to_path
    parameters:
      out-path: data/sorted_countries_by_name
Explanation: Pipeline Dependencies
Let's add another pipeline which depends on the countries-population pipeline.
This time we will use just the pipeline spec yaml to write the pipeline, without any DataFlows code (although DataFlows library is used to implement the processors we are using here):
End of explanation
%%sh
dpp init
dpp
Explanation: Clear the pipelines state using dpp init and list the available pipelines:
End of explanation
%%sh
dpp run --dirty all
Explanation: You can see that the new pipeline can't run until it's dependency is executed.
Let's run all the "Dirty" dependencies:
End of explanation
from dataflows import Flow, load, printer
Flow(
    load('data/sorted_countries_by_name/datapackage.json'),
    printer(num_rows=1, tablefmt='html')
).process()
Explanation: Inspect the created datapackage
End of explanation
%%writefile pipeline-spec.yaml
double-winners:
  pipeline:
  - run: load
    parameters:
      name: emmies
      from: https://raw.githubusercontent.com/datahq/dataflows/master/data/emmy.csv
  - run: load
    parameters:
      name: oscars
      from: https://raw.githubusercontent.com/datahq/dataflows/master/data/academy.csv
  - run: filter
    parameters:
      resources: ['emmies']
      in:
      - winner: 1
  - run: concatenate
    parameters:
      target: {'name': 'emmies_filtered'}
      resources: ['emmies']
      fields:
        emmy_nominee: ['nominee']
  - run: join
    parameters:
      source:
        name: 'emmies_filtered'
        key: ['emmy_nominee']
        delete: true
      target:
        name: 'oscars'
        key: ['Name']
      fields: {}
      full: false
  - run: filter
    parameters:
      in:
      - Winner: "1"
  - run: dump.to_path
    parameters:
      out-path: data/double_winners
Explanation: Pipeline processors
Datapackage Pipelines has a standard library of processors, like the sort processor used previously. These processors correspond to DataFlows standard library processors.
See the Datapackage Pipelines README for reference and usage examples.
An example showing usage of common processors:
End of explanation
%%sh
dpp run ./double-winners
Explanation: Run the pipeline:
End of explanation
from dataflows import Flow, printer, load
Flow(load('data/double_winners/datapackage.json'), printer(tablefmt='html', num_rows=1)).process()
Explanation: Print the datapackage:
End of explanation
%%sh
docker pull frictionlessdata/datapackage-pipelines
Explanation: Pipelines Server
Running pipelines on your laptop is fine for many use-cases but sometimes you want to run pipelines in a more reproducible, scalable and automatic fashion.
The Datapackage Pipelines Server is a Docker image which provides the core functionality to achieve this.
To start a local pipelines server for development, you will need to install Docker for Windows,
Mac or Linux
Pull the datapackage-pipelines image:
End of explanation
%%sh
docker run -d --name dpp -v `pwd`:/pipelines:rw -p 5000:5000 frictionlessdata/datapackage-pipelines server
Explanation: Start a local pipelines server, mounting the current working directory into the container:
End of explanation
%%writefile pipeline-spec.yaml
countries-population:
  schedule:
    # minute hour day_of_week day_of_month month_of_year
    crontab: '* * * * *'
  pipeline:
  - flow: countries_population_flow
  - run: dump.to_path
    parameters:
      out-path: data/countries_population
sorted_countries_by_name:
  dependencies:
  - pipeline: ./countries-population
  - datapackage: data/countries_population/datapackage.json
  pipeline:
  - run: load
    parameters:
      from: data/countries_population/datapackage.json
      resources: ['countries_population']
  - run: sort
    parameters:
      resources: ['countries_population']
      sort-by: '{name}'
  - run: dump.to_path
    parameters:
      out-path: data/sorted_countries_by_name
Explanation: After a few seconds, the pipelines dashboad should be available at http://localhost:5000
New / modified pipelines and dirty dependencies are executed by the pipelines server automatically.
The server also supports scheduled pipelines for periodical execution.
Let's see this in action:
End of explanation
%%sh
docker logs dpp --tail 5
Explanation: Inspect the Pipelines server logs and wait for Update Pipelines task to complete and pipelines to start running
End of explanation
<END_TASK>