Example Pipeline

Our first task will be to convert a ZIP code TSV into a set of county level entries.

The input file looks like:

ZIP,COUNTYNAME,STATE,STCOUNTYFP,CLASSFP
36003,Autauga County,AL,01001,H1
36006,Autauga County,AL,01001,H1
36067,Autauga County,AL,01001,H1
36066,Autauga County,AL,01001,H1
36703,Autauga County,AL,01001,H1
36701,Autauga County,AL,01001,H1
36091,Autauga County,AL,01001,H1

First is the header of the pipeline. This declares the unique name of the pipeline and it’s output directory.

name: zipcode_map
outdir: ./
docs: Converts zipcode TSV into graph elements

Next the configuration is declared. In this case the only input is the zipcode TSV. There is a default value, so the pipeline can be invoked without passing in any parameters. However, to apply this pipeline to a new input file, the input parameter zipcode could be used to define the source file.

config:
  schema: ../covid19_datadictionary/gdcdictionary/schemas/
  zipcode: ../data/ZIP-COUNTY-FIPS_2017-06.csv

The inputs section declares data input sources. In this pipeline, there is only one input, which is to run the table loader.

inputs:
  tableLoad:
    input: "{{config.zipcode}}"
    sep: ","

Tableload operaters of the input file that was originally passed in using the inputs stanza. SIFTER string parsing is based on mustache template system. To access the string passed in the template is {{config.zipcode}}. The seperator in the file input file is a , so that is also passed in as a parameter to the extractor.

The tableLoad extractor opens up the TSV and generates a one message for every row in the file. It uses the header of the file to map the column values into a dictionary. The first row would produce the message:

{
    "ZIP" : "36003",
    "COUNTYNAME" : "Autauga County",
    "STATE" : "AL",
    "STCOUNTYFP" : "01001",
    "CLASSFP" : "H1"
}

The stream of messages are then passed into the steps listed in the transform section of the tableLoad extractor.

For the current tranform, we want to produce a single entry per STCOUNTYFP, however, the file has a line per ZIP. We need to run a reduce transform, that collects rows togeather using a field key, which in this case is "{{row.STCOUNTYFP}}", and then runs a function merge that takes two messages, merges them togeather and produces a single output message.

The two messages:

{ "ZIP" : "36003", "COUNTYNAME" : "Autauga County", "STATE" : "AL", "STCOUNTYFP" : "01001", "CLASSFP" : "H1"}
{ "ZIP" : "36006", "COUNTYNAME" : "Autauga County", "STATE" : "AL", "STCOUNTYFP" : "01001", "CLASSFP" : "H1"}

Would be merged into the message:

{ "ZIP" : ["36003", "36006"], "COUNTYNAME" : "Autauga County", "STATE" : "AL", "STCOUNTYFP" : "01001", "CLASSFP" : "H1"}

The reduce transform step uses a block of python code to describe the function. The method field names the function, in this case merge that will be used as the reduce function.

  zipReduce:
    - from: zipcode
    - reduce:
        field: STCOUNTYFP
        method: merge
        python: >
          def merge(x,y):
            a = x.get('zipcodes', []) + [x['ZIP']]
            b = y.get('zipcodes', []) + [y['ZIP']]
            x['zipcodes'] = a + b
            return x

The original messages produced by the loader have all of the information required by the summary_location object type as described by the JSON schema that was linked to in the header stanza. However, the data is all under the wrong field names. To remap the data, we use a project tranformation that uses the template engine to project data into new files in the message. The template engine has the current message data in the value row. So the value FIPS:{{row.STCOUNTYFP}} is mapped into the field id.

  - project:
      mapping:
        id: "FIPS:{{row.STCOUNTYFP}}"
        province_state: "{{row.STATE}}"
        summary_locations: "{{row.STCOUNTYFP}}"
        county: "{{row.COUNTYNAME}}"
        submitter_id: "{{row.STCOUNTYFP}}"
        type: summary_location
        projects: []

Using this projection, the message:

{
  "ZIP" : ["36003", "36006"],
  "COUNTYNAME" : "Autauga County",
  "STATE" : "AL",
  "STCOUNTYFP" : "01001",
  "CLASSFP" : "H1"
}

would become

{
  "id" : "FIPS:01001",
  "province_state" : "AL",
  "summary_locations" : "01001",
  "county" : "Autauga County",
  "submitter_id" : "01001",
  "type" : "summary_location"
  "projects" : [],
  "ZIP" : ["36003", "36006"],
  "COUNTYNAME" : "Autauga County",
  "STATE" : "AL",
  "STCOUNTYFP" : "01001",
  "CLASSFP" : "H1"
}

Now that the data has been remapped, we pass the data into the ‘objectCreate’ transformation, which will read in the schema for summary_location, check the message to make sure it matches and then output it.

  - objectCreate:
        class: summary_location

Outputs

To create an output table, with two columns connecting ZIP values to STCOUNTYFP values. The STCOUNTYFP is a county level FIPS code, used by the census office. A single FIPS code my contain many ZIP codes, and we can use this table later for mapping ids when loading the data into a database.

outputs:
  zip2fips:
    tableWrite:
      from: 
      output: zip2fips
      columns:
        - ZIP
        - STCOUNTYFP

Sifter

Example Pipeline