examples/multi_input_output.yaml

pipeline:
  type: composite
  transforms:

    # Create some input.
    - type: Create
      name: Weather
      config:
        elements:
          - {date: '2024-12-01', city: 'Seattle', temperature: 51}
          - {date: '2024-12-02', city: 'Seattle', temperature: 51}
          - {date: '2024-12-03', city: 'Seattle', temperature: 45}
          - {date: '2024-12-03', city: 'Kirkland', temperature: 52}
          - {date: '2024-12-01', city: 'New York City', temperature: 41}
          - {date: '2024-12-02', city: 'New York City', temperature: 43}
          
    - type: Create
      name: CityToState
      config:
        elements:
          - {city: 'Seattle', state: 'WA'}
          - {city: 'Kirkland', state: 'WA'}
          - {city: 'New York City', state: 'NY'}

    - type: MultiInputMultiOutput
      input:
        # Specify multiple inputs as a mapping that will get passed as a dict
        # to the transform's expand function.
        city_measurements: Weather
        city_to_state: CityToState
    
    # Multiple outputs are returned from the expand function as a dictionary
    # mapping string keys to PCollections. Those string keys are referenced
    # here.
    - type: LogForTesting
      input: MultiInputMultiOutput.avg_per_city

    - type: LogForTesting
      input: MultiInputMultiOutput.max_per_state
        
    # Ensure that the pipeline ran correctly.
    - type: AssertEqual
      input: MultiInputMultiOutput.avg_per_city
      config:
        elements:
          - {city: 'Seattle', temperature: 49}
          - {city: 'Kirkland', temperature: 52}
          - {city: 'New York City', temperature: 42}
        
    # Ensure that the pipeline ran correctly.
    - type: AssertEqual
      input: MultiInputMultiOutput.max_per_state
      config:
        elements:
          - {state: 'WA', temperature: 52}
          - {state: 'NY', temperature: 43}

providers:
  - include: "./provider_listing.yaml"