New York Taxi Data

    • import from raw data
    • download of prepared partitions

    See https://github.com/toddwschneider/nyc-taxi-data and for the description of a dataset and instructions for downloading.

    Downloading will result in about 227 GB of uncompressed data in CSV files. The download takes about an hour over a 1 Gbit connection (parallel downloading from s3.amazonaws.com recovers at least half of a 1 Gbit channel).
    Some of the files might not download fully. Check the file sizes and re-download any that seem doubtful.

    Some of the files might contain invalid rows. You can fix them as follows:

    Then the data must be pre-processed in PostgreSQL. This will create selections of points in the polygons (to match points on the map with the boroughs of New York City) and combine all the data into a single denormalized flat table by using a JOIN. To do this, you will need to install PostgreSQL with PostGIS support.

    Be careful when running and manually re-check that all the tables were created correctly.

    It takes about 20-30 minutes to process each month’s worth of data in PostgreSQL, for a total of about 48 hours.

    You can check the number of downloaded rows as follows:

    1. $ time psql nyc-taxi-data -c "SELECT count(*) FROM trips;"
    2. ## Count
    3.  1298979494
    4. (1 row)
    6. real    7m9.164s

    (This is slightly more than 1.1 billion rows reported by Mark Litwintschik in a series of blog posts.)

    The data in PostgreSQL uses 370 GB of space.

    Exporting the data from PostgreSQL:

    1. COPY
    2. (
    3.     SELECT trips.id,
    4.            trips.vendor_id,
    5.            trips.pickup_datetime,
    6.            trips.dropoff_datetime,
    7.            trips.store_and_fwd_flag,
    8.            trips.rate_code_id,
    9.            trips.pickup_longitude,
    10.            trips.pickup_latitude,
    11.            trips.dropoff_longitude,
    12.            trips.dropoff_latitude,
    13.            trips.passenger_count,
    14.            trips.trip_distance,
    15.            trips.fare_amount,
    16.            trips.extra,
    17.            trips.mta_tax,
    18.            trips.tip_amount,
    19.            trips.tolls_amount,
    20.            trips.ehail_fee,
    21.            trips.improvement_surcharge,
    22.            trips.total_amount,
    23.            trips.payment_type,
    24.            trips.trip_type,
    25.            trips.pickup,
    26.            trips.dropoff,
    28.            cab_types.type cab_type,
    30.            weather.precipitation_tenths_of_mm rain,
    31.            weather.snow_depth_mm,
    32.            weather.snowfall_mm,
    33.            weather.max_temperature_tenths_degrees_celsius max_temp,
    34.            weather.min_temperature_tenths_degrees_celsius min_temp,
    35.            weather.average_wind_speed_tenths_of_meters_per_second wind,
    37.            pick_up.gid pickup_nyct2010_gid,
    38.            pick_up.ctlabel pickup_ctlabel,
    39.            pick_up.boroname pickup_boroname,
    40.            pick_up.ct2010 pickup_ct2010,
    41.            pick_up.boroct2010 pickup_boroct2010,
    42.            pick_up.cdeligibil pickup_cdeligibil,
    44.            pick_up.ntaname pickup_ntaname,
    45.            pick_up.puma pickup_puma,
    47.            drop_off.gid dropoff_nyct2010_gid,
    48.            drop_off.ctlabel dropoff_ctlabel,
    49.            drop_off.borocode dropoff_borocode,
    50.            drop_off.boroname dropoff_boroname,
    51.            drop_off.ct2010 dropoff_ct2010,
    52.            drop_off.boroct2010 dropoff_boroct2010,
    53.            drop_off.cdeligibil dropoff_cdeligibil,
    54.            drop_off.ntacode dropoff_ntacode,
    55.            drop_off.ntaname dropoff_ntaname,
    56.            drop_off.puma dropoff_puma
    57.     FROM trips
    58.     LEFT JOIN cab_types
    59.         ON trips.cab_type_id = cab_types.id
    60.     LEFT JOIN central_park_weather_observations_raw weather
    61.         ON weather.date = trips.pickup_datetime::date
    62.     LEFT JOIN nyct2010 pick_up
    63.         ON pick_up.gid = trips.pickup_nyct2010_gid
    64.     LEFT JOIN nyct2010 drop_off
    65.         ON drop_off.gid = trips.dropoff_nyct2010_gid
    66. ) TO '/opt/milovidov/nyc-taxi-data/trips.tsv';

    The data snapshot is created at a speed of about 50 MB per second. While creating the snapshot, PostgreSQL reads from the disk at a speed of about 28 MB per second.
    This takes about 5 hours. The resulting TSV file is 590612904969 bytes.

    Create a temporary table in ClickHouse:

    1. CREATE TABLE trips
    2. (
    3. trip_id                 UInt32,
    4. vendor_id               String,
    5. pickup_datetime         DateTime,
    6. dropoff_datetime        Nullable(DateTime),
    7. store_and_fwd_flag      Nullable(FixedString(1)),
    8. rate_code_id            Nullable(UInt8),
    9. pickup_longitude        Nullable(Float64),
    10. pickup_latitude         Nullable(Float64),
    11. dropoff_longitude       Nullable(Float64),
    12. dropoff_latitude        Nullable(Float64),
    13. passenger_count         Nullable(UInt8),
    14. trip_distance           Nullable(Float64),
    15. fare_amount             Nullable(Float32),
    16. extra                   Nullable(Float32),
    17. mta_tax                 Nullable(Float32),
    18. tip_amount              Nullable(Float32),
    19. tolls_amount            Nullable(Float32),
    20. ehail_fee               Nullable(Float32),
    21. improvement_surcharge   Nullable(Float32),
    22. total_amount            Nullable(Float32),
    23. trip_type               Nullable(UInt8),
    24. pickup                  Nullable(String),
    25. dropoff                 Nullable(String),
    27. precipitation           Nullable(UInt8),
    28. snow_depth              Nullable(UInt8),
    29. snowfall                Nullable(UInt8),
    30. max_temperature         Nullable(UInt8),
    31. min_temperature         Nullable(UInt8),
    32. average_wind_speed      Nullable(UInt8),
    33. pickup_nyct2010_gid     Nullable(UInt8),
    34. pickup_ctlabel          Nullable(String),
    35. pickup_borocode         Nullable(UInt8),
    36. pickup_boroname         Nullable(String),
    37. pickup_ct2010           Nullable(String),
    38. pickup_boroct2010       Nullable(String),
    39. pickup_cdeligibil       Nullable(FixedString(1)),
    40. pickup_ntacode          Nullable(String),
    41. pickup_ntaname          Nullable(String),
    42. pickup_puma             Nullable(String),
    43. dropoff_nyct2010_gid    Nullable(UInt8),
    44. dropoff_ctlabel         Nullable(String),
    45. dropoff_borocode        Nullable(UInt8),
    46. dropoff_boroname        Nullable(String),
    47. dropoff_ct2010          Nullable(String),
    48. dropoff_boroct2010      Nullable(String),
    49. dropoff_cdeligibil      Nullable(String),
    50. dropoff_ntacode         Nullable(String),
    51. dropoff_ntaname         Nullable(String),
    52. dropoff_puma            Nullable(String)
    53. ) ENGINE = Log;

    It is needed for converting fields to more correct data types and, if possible, to eliminate NULLs.

    1. $ time clickhouse-client --query="INSERT INTO trips FORMAT TabSeparated" < trips.tsv
    3. real    75m56.214s

    Data is read at a speed of 112-140 Mb/second.
    Loading data into a Log type table in one stream took 76 minutes.
    The data in this table uses 142 GB.

    Unfortunately, all the fields associated with the weather (precipitation…average_wind_speed) were filled with NULL. Because of this, we will remove them from the final data set.

    To start, we’ll create a table on a single server. Later we will make the table distributed.

    Create and populate a summary table:

    This takes 3030 seconds at a speed of about 428,000 rows per second.
    To load it faster, you can create the table with the Log engine instead of MergeTree. In this case, the download works faster than 200 seconds.

    The table uses 126 GB of disk space.

    1. SELECT formatReadableSize(sum(bytes)) FROM system.parts WHERE table = 'trips_mergetree' AND active
    1. ┌─formatReadableSize(sum(bytes))─┐
    2. │ 126.18 GiB                     │
    3. └────────────────────────────────┘

    Among other things, you can run the OPTIMIZE query on MergeTree. But it’s not required since everything will be fine without it.

    1. $ curl -O https://datasets.clickhouse.tech/trips_mergetree/partitions/trips_mergetree.tar
    2. $ tar xvf trips_mergetree.tar -C /var/lib/clickhouse # path to ClickHouse data directory
    3. $ # check permissions of unpacked data, fix if required
    4. $ sudo service clickhouse-server restart
    5. $ clickhouse-client --query "select count(*) from datasets.trips_mergetree"

    Info

    If you will run the queries described below, you have to use the full table name, datasets.trips_mergetree.

    Q1:

    1. SELECT cab_type, count(*) FROM trips_mergetree GROUP BY cab_type

    0.490 seconds.

    Q2:

    1.224 seconds.

    Q3:

    1. SELECT passenger_count, toYear(pickup_date) AS year, count(*) FROM trips_mergetree GROUP BY passenger_count, year

    2.104 seconds.

    1. SELECT passenger_count, toYear(pickup_date) AS year, round(trip_distance) AS distance, count(*)
    2. FROM trips_mergetree
    3. GROUP BY passenger_count, year, distance
    4. ORDER BY year, count(*) DESC

    3.593 seconds.

    The following server was used:

    Two Intel(R) Xeon(R) CPU E5-2650 v2 @ 2.60GHz, 16 physical kernels total,128 GiB RAM,8x6 TB HD on hardware RAID-5

    Execution time is the best of three runs. But starting from the second run, queries read data from the file system cache. No further caching occurs: the data is read out and processed in each run.

    Creating a table on three servers:

    On each server:

    On the source server:

      The following query redistributes data:

      This takes 2454 seconds.

      On three servers:

      Q1: 0.212 seconds.
      Q2: 0.438 seconds.
      Q3: 0.733 seconds.
      Q4: 1.241 seconds.

      No surprises here, since the queries are scaled linearly.

      We also have the results from a cluster of 140 servers:

      Q1: 0.028 sec.
      Q2: 0.043 sec.
      Q3: 0.051 sec.
      Q4: 0.072 sec.

      Original article