Snowplow BigQuery Loader

This project contains applications used to load Snowplow enriched data into Google BigQuery.

Quickstart

Assuming git and SBT installed:

$ git clone https://github.com/snowplow-incubator/snowplow-bigquery-loader
$ cd snowplow-bigquery-loader
$ sbt "project loader" test
$ sbt "project mutator" test

Deduplication steps on BigQuery

Before running these queires, duplicates dateset should be deleted if exists and created again after that. After configure dataset we need to run below sql script to save duplicated event ids in another table.

CREATE TABLE duplicates.tmp_events_id
AS (

  SELECT event_id
  FROM (

    SELECT event_id, COUNT(*) AS count
    FROM prisma_dataset.events
      --AND collector_tstamp > DATEADD(week, -4, CURRENT_DATE) -- restricts table scan for the previous 4 weeks to make queries more efficient; uncomment after running the first time
    GROUP BY 1

  )

  WHERE count > 1

);

We have duplicated event ids. We also need to save duplicated event rows with row numbers in another table. Because in the next step we'll delete all these events from events table and add just first row of each duplicated event.

CREATE TABLE duplicates.tmp_events
AS (

  SELECT *, ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY dvce_created_tstamp) as event_number
  FROM prisma_dataset.events
  WHERE event_id IN (SELECT event_id FROM duplicates.tmp_events_id)

);

Now we can delete these events

DELETE FROM prisma_dataset.events
  WHERE event_id IN (SELECT event_id FROM duplicates.tmp_events_id) AND collector_tstamp < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 40 MINUTE);

We need to use this where condition because BQ doesn't allow deleting of just streamed rows. Or we need to shut down collector and after wait a while we can run the query without this time condition in where state.

We have clean table now but we lost events which was duplicated. We can add just first line of them with the query below.

INSERT INTO prisma_dataset.events (

    SELECT

      app_id, platform, etl_tstamp, collector_tstamp, dvce_created_tstamp, event, event_id, txn_id,
      name_tracker, v_tracker, v_collector, v_etl,
      user_id, user_ipaddress, user_fingerprint, domain_userid, domain_sessionidx, network_userid,
      geo_country, geo_region, geo_city, geo_zipcode, geo_latitude, geo_longitude, geo_region_name,
      ip_isp, ip_organization, ip_domain, ip_netspeed, page_url, page_title, page_referrer,
      page_urlscheme, page_urlhost, page_urlport, page_urlpath, page_urlquery, page_urlfragment,
      refr_urlscheme, refr_urlhost, refr_urlport, refr_urlpath, refr_urlquery, refr_urlfragment,
      refr_medium, refr_source, refr_term, mkt_medium, mkt_source, mkt_term, mkt_content, mkt_campaign,
      se_category, se_action, se_label, se_property, se_value,
      tr_orderid, tr_affiliation, tr_total, tr_tax, tr_shipping, tr_city, tr_state, tr_country,
      ti_orderid, ti_sku, ti_name, ti_category, ti_price, ti_quantity,
      pp_xoffset_min, pp_xoffset_max, pp_yoffset_min, pp_yoffset_max,
      useragent, br_name, br_family, br_version, br_type, br_renderengine, br_lang, br_features_pdf, br_features_flash,
      br_features_java, br_features_director, br_features_quicktime, br_features_realplayer, br_features_windowsmedia,
      br_features_gears, br_features_silverlight, br_cookies, br_colordepth, br_viewwidth, br_viewheight,
      os_name, os_family, os_manufacturer, os_timezone, dvce_type, dvce_ismobile, dvce_screenwidth, dvce_screenheight,
      doc_charset, doc_width, doc_height, tr_currency, tr_total_base, tr_tax_base, tr_shipping_base,
      ti_currency, ti_price_base, base_currency, geo_timezone, mkt_clickid, mkt_network, etl_tags,
      dvce_sent_tstamp, refr_domain_userid, refr_dvce_tstamp, domain_sessionid,
      derived_tstamp, event_vendor, event_name, event_format, event_version, event_fingerprint, true_tstamp,
      contexts_com_snowplowanalytics_snowplow_mobile_context_1_0_1, contexts_com_snowplowanalytics_snowplow_client_session_1_0_1, unstruct_event_com_snowplowanalytics_snowplow_screen_view_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_error_1_0_0, contexts_com_snowplowanalytics_snowplow_geolocation_context_1_1_0, unstruct_event_com_snowplowanalytics_snowplow_timing_1_0_0, contexts_com_snowplowanalytics_mobile_application_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_background_1_0_0, contexts_com_snowplowanalytics_mobile_screen_1_0_0, unstruct_event_com_snowplowanalytics_mobile_screen_view_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_link_click_1_0_1, unstruct_event_com_snowplowanalytics_mobile_application_install_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_foreground_1_0_0

    FROM duplicates.tmp_events WHERE event_number = 1

  );

Deduplication steps on BigQuery for duplicated records

Before running these queires, duplicate_structs dateset should be deleted if exists and created again after that. After configure dataset we need to run below sql script to save duplicated event ids in another table.

CREATE TABLE duplicate_structs.tmp_events_id
AS (

  SELECT event_id
  FROM (

    SELECT event_id, ARRAY_LENGTH( contexts_com_snowplowanalytics_snowplow_client_session_1_0_1 ) AS count
    FROM prisma_dataset.events
    WHERE app_id != 'co.fourapps.aword'

  )

  WHERE count > 1

);

We have event ids of events which has duplicated records. We can move these events to another table with just first record of duplicated records.

CREATE TABLE duplicate_structs.tmp_events
AS (

  SELECT
  app_id, platform, etl_tstamp, collector_tstamp, dvce_created_tstamp, event, event_id, txn_id,
      name_tracker, v_tracker, v_collector, v_etl,
      user_id, user_ipaddress, user_fingerprint, domain_userid, domain_sessionidx, network_userid,
      geo_country, geo_region, geo_city, geo_zipcode, geo_latitude, geo_longitude, geo_region_name,
      ip_isp, ip_organization, ip_domain, ip_netspeed, page_url, page_title, page_referrer,
      page_urlscheme, page_urlhost, page_urlport, page_urlpath, page_urlquery, page_urlfragment,
      refr_urlscheme, refr_urlhost, refr_urlport, refr_urlpath, refr_urlquery, refr_urlfragment,
      refr_medium, refr_source, refr_term, mkt_medium, mkt_source, mkt_term, mkt_content, mkt_campaign,
      se_category, se_action, se_label, se_property, se_value,
      tr_orderid, tr_affiliation, tr_total, tr_tax, tr_shipping, tr_city, tr_state, tr_country,
      ti_orderid, ti_sku, ti_name, ti_category, ti_price, ti_quantity,
      pp_xoffset_min, pp_xoffset_max, pp_yoffset_min, pp_yoffset_max,
      useragent, br_name, br_family, br_version, br_type, br_renderengine, br_lang, br_features_pdf, br_features_flash,
      br_features_java, br_features_director, br_features_quicktime, br_features_realplayer, br_features_windowsmedia,
      br_features_gears, br_features_silverlight, br_cookies, br_colordepth, br_viewwidth, br_viewheight,
      os_name, os_family, os_manufacturer, os_timezone, dvce_type, dvce_ismobile, dvce_screenwidth, dvce_screenheight,
      doc_charset, doc_width, doc_height, tr_currency, tr_total_base, tr_tax_base, tr_shipping_base,
      ti_currency, ti_price_base, base_currency, geo_timezone, mkt_clickid, mkt_network, etl_tags,
      dvce_sent_tstamp, refr_domain_userid, refr_dvce_tstamp, domain_sessionid,
      derived_tstamp, event_vendor, event_name, event_format, event_version, event_fingerprint, true_tstamp,
      contexts_com_snowplowanalytics_snowplow_mobile_context_1_0_1, contexts_com_snowplowanalytics_snowplow_client_session_1_0_1[OFFSET(0)] AS contexts_com_snowplowanalytics_snowplow_client_session_1_0_1, unstruct_event_com_snowplowanalytics_snowplow_screen_view_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_error_1_0_0, contexts_com_snowplowanalytics_snowplow_geolocation_context_1_1_0, unstruct_event_com_snowplowanalytics_snowplow_timing_1_0_0, contexts_com_snowplowanalytics_mobile_application_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_background_1_0_0, contexts_com_snowplowanalytics_mobile_screen_1_0_0, unstruct_event_com_snowplowanalytics_mobile_screen_view_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_link_click_1_0_1, unstruct_event_com_snowplowanalytics_mobile_application_install_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_foreground_1_0_0
  FROM prisma_dataset.events
  WHERE event_id IN (SELECT event_id FROM duplicate_structs.tmp_events_id) AND app_id != 'co.fourapps.aword'

);

Delete these events from events table.

  DELETE FROM prisma_dataset.events
  WHERE event_id IN (SELECT event_id FROM duplicate_structs.tmp_events_id) AND collector_tstamp < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 40 MINUTE)
  ;

After that add these events with clean format again.

INSERT INTO prisma_dataset.events (SELECT

app_id, platform, etl_tstamp, collector_tstamp, dvce_created_tstamp, event, event_id, txn_id,
      name_tracker, v_tracker, v_collector, v_etl,
      user_id, user_ipaddress, user_fingerprint, domain_userid, domain_sessionidx, network_userid,
      geo_country, geo_region, geo_city, geo_zipcode, geo_latitude, geo_longitude, geo_region_name,
      ip_isp, ip_organization, ip_domain, ip_netspeed, page_url, page_title, page_referrer,
      page_urlscheme, page_urlhost, page_urlport, page_urlpath, page_urlquery, page_urlfragment,
      refr_urlscheme, refr_urlhost, refr_urlport, refr_urlpath, refr_urlquery, refr_urlfragment,
      refr_medium, refr_source, refr_term, mkt_medium, mkt_source, mkt_term, mkt_content, mkt_campaign,
      se_category, se_action, se_label, se_property, se_value,
      tr_orderid, tr_affiliation, tr_total, tr_tax, tr_shipping, tr_city, tr_state, tr_country,
      ti_orderid, ti_sku, ti_name, ti_category, ti_price, ti_quantity,
      pp_xoffset_min, pp_xoffset_max, pp_yoffset_min, pp_yoffset_max,
      useragent, br_name, br_family, br_version, br_type, br_renderengine, br_lang, br_features_pdf, br_features_flash,
      br_features_java, br_features_director, br_features_quicktime, br_features_realplayer, br_features_windowsmedia,
      br_features_gears, br_features_silverlight, br_cookies, br_colordepth, br_viewwidth, br_viewheight,
      os_name, os_family, os_manufacturer, os_timezone, dvce_type, dvce_ismobile, dvce_screenwidth, dvce_screenheight,
      doc_charset, doc_width, doc_height, tr_currency, tr_total_base, tr_tax_base, tr_shipping_base,
      ti_currency, ti_price_base, base_currency, geo_timezone, mkt_clickid, mkt_network, etl_tags,
      dvce_sent_tstamp, refr_domain_userid, refr_dvce_tstamp, domain_sessionid,
      derived_tstamp, event_vendor, event_name, event_format, event_version, event_fingerprint, true_tstamp,
      contexts_com_snowplowanalytics_snowplow_mobile_context_1_0_1, ARRAY<STRUCT<session_id STRING, session_index INT64, storage_mechanism STRING, user_id STRING, first_event_id STRING, previous_session_id STRING>>[contexts_com_snowplowanalytics_snowplow_client_session_1_0_1] AS contexts_com_snowplowanalytics_snowplow_client_session_1_0_1, unstruct_event_com_snowplowanalytics_snowplow_screen_view_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_error_1_0_0, contexts_com_snowplowanalytics_snowplow_geolocation_context_1_1_0, unstruct_event_com_snowplowanalytics_snowplow_timing_1_0_0, contexts_com_snowplowanalytics_mobile_application_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_background_1_0_0, contexts_com_snowplowanalytics_mobile_screen_1_0_0, unstruct_event_com_snowplowanalytics_mobile_screen_view_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_link_click_1_0_1, unstruct_event_com_snowplowanalytics_mobile_application_install_1_0_0, unstruct_event_com_snowplowanalytics_snowplow_application_foreground_1_0_0

FROM duplicate_structs.tmp_events)

Copyright and License

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this software except in compliance with the License.

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

kayalardanmehmet / snowplow-bigquery-loader Goto Github PK