The NycTaxiZipFareExtractor operator is the operator that is immediate downstream of the NycTaxiCsvParser explained above. It looks at the pickup latitude-longitude coordinate, looks up the zip code given the latitude-longitude coordinate, and prepares a TimestampedTuple for the KeyedWindowedOperatorImpl operator downstream. The output tuple is of the TimestampedTuple<KeyValPair<String, Double>> type, with the key being the zip code, and the value being the total payment amount.
Here's the definition of the input port of the NycTaxiZipFareExtractor operator:
public final transient DefaultInputPort<Map<String, String>> input = new DefaultInputPort<Map<String, String>>()
{
@Override
public void process(Map<String, String> tuple)
{
try {
String zip =
NycLocationUtils.getZip(Double.valueOf(tuple.get("pickup_lat")),
Double.valueOf(tuple.get("pickup_lon...