package com.sony.etl; import com.google.api.services.bigquery.model.TableFieldSchema; import com.google.api.services.bigquery.model.TableRow; import com.google.api.services.bigquery.model.TableSchema; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.io.TextIO; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Description; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.ParDo; import org.apache.beam.sdk.values.PCollection; import org.json.JSONObject; import java.util.ArrayList; import java.util.Iterator; import java.util.List; public class SonyAggstreamstoBQ { public interface Options extends PipelineOptions { @Description("INPUT File") @Default.String("/test") ValueProvider getInputFile(); void setInputFile(ValueProvider value); @Description("OUTPUT BigQuery Dataset") ValueProvider getOutputDataset(); void setOutputDataset(ValueProvider value); } private static TableSchema createBigQuerySchema() { List fields = new ArrayList<>(); fields.add(new TableFieldSchema().setName("album_href").setType("STRING")); fields.add(new TableFieldSchema().setName("album_name").setType("STRING")); fields.add(new TableFieldSchema().setName("artists_hrefs").setType("STRING")); fields.add(new TableFieldSchema().setName("artists_names").setType("STRING")); fields.add(new TableFieldSchema().setName("country").setType("STRING")); fields.add(new TableFieldSchema().setName("date").setType("DATE")); fields.add(new TableFieldSchema().setName("label").setType("STRING")); fields.add(new TableFieldSchema().setName("licensor").setType("STRING")); fields.add(new TableFieldSchema().setName("message_name").setType("STRING")); fields.add(new TableFieldSchema().setName("saves_product_free").setType("INT64")); fields.add(new TableFieldSchema().setName("saves_product_premium").setType("INT64")); fields.add(new TableFieldSchema().setName("saves").setType("INT64")); fields.add(new TableFieldSchema().setName("skips").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_0_17").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_18_22").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_23_27").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_28_34").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_35_44").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_45_59").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_60_150").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_female_age_unknown").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_0_17").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_18_22").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_23_27").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_28_34").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_35_44").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_45_59").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_60_150").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_male_age_unknown").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_0_17").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_18_22").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_23_27").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_28_34").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_35_44").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_45_59").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_60_150").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_neutral_age_unknown").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_0_17").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_18_22").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_23_27").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_28_34").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_35_44").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_45_59").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_60_150").setType("INT64")); fields.add(new TableFieldSchema().setName("streams_sex_unknown_age_unknown").setType("INT64")); fields.add(new TableFieldSchema().setName("streams").setType("INT64")); fields.add(new TableFieldSchema().setName("track_href").setType("STRING")); fields.add(new TableFieldSchema().setName("track_isrc").setType("STRING")); fields.add(new TableFieldSchema().setName("track_name").setType("STRING")); fields.add(new TableFieldSchema().setName("version").setType("STRING")); return new TableSchema().setFields(fields); } public static class FormatBigQueryFn extends DoFn { @ProcessElement public void processElement(ProcessContext c, @Element String data, OutputReceiver out){ List rows = new ArrayList<>(); JSONObject obj = new JSONObject(data); int saves_product_free = 0; int saves_product_premium = 0; int streams_sex_female_age_0_17 = 0; int streams_sex_female_age_18_22 = 0; int streams_sex_female_age_23_27 = 0; int streams_sex_female_age_28_34 = 0; int streams_sex_female_age_35_44 = 0; int streams_sex_female_age_45_59 = 0; int streams_sex_female_age_60_150 = 0; int streams_sex_female_age_unknown = 0; int streams_sex_male_age_0_17 = 0; int streams_sex_male_age_18_22 = 0; int streams_sex_male_age_23_27 = 0; int streams_sex_male_age_28_34 = 0; int streams_sex_male_age_35_44 = 0; int streams_sex_male_age_45_59 = 0; int streams_sex_male_age_60_150 = 0; int streams_sex_male_age_unknown = 0; int streams_sex_neutral_age_0_17 = 0; int streams_sex_neutral_age_18_22 = 0; int streams_sex_neutral_age_23_27 = 0; int streams_sex_neutral_age_28_34 = 0; int streams_sex_neutral_age_35_44 = 0; int streams_sex_neutral_age_45_59 = 0; int streams_sex_neutral_age_60_150 = 0; int streams_sex_neutral_age_unknown = 0; int streams_sex_unknown_age_0_17 = 0; int streams_sex_unknown_age_18_22 = 0; int streams_sex_unknown_age_23_27 = 0; int streams_sex_unknown_age_28_34 = 0; int streams_sex_unknown_age_35_44 = 0; int streams_sex_unknown_age_45_59 = 0; int streams_sex_unknown_age_60_150 = 0; int streams_sex_unknown_age_unknown = 0; int saves_total = 0; int streams_total = 0; String licensor = (String) obj.get("licensor"); String label = (String) obj.get("label"); String version = (String) obj.get("version"); String date = (String) obj.get("date"); String message_name = (String) obj.get("message_name"); String artists_names = (String) ((JSONObject) obj.get("artists")).get("names"); String artists_hrefs = (String) ((JSONObject) obj.get("artists")).get("hrefs"); String album_name = (String) ((JSONObject) obj.get("album")).get("name"); String album_href = (String) ((JSONObject) obj.get("album")).get("href"); String track_name = (String) ((JSONObject) obj.get("trackv2")).get("name"); String track_href = (String) ((JSONObject) obj.get("trackv2")).get("href"); String track_isrc = (String) ((JSONObject) obj.get("trackv2")).get("isrc"); int skips = (int) ((JSONObject) obj.get("skips")).get("total"); JSONObject saves = ((JSONObject) ((JSONObject) obj.get("saves")).get("country")); JSONObject streams = ((JSONObject) ((JSONObject) obj.get("streams")).get("country")); for (Iterator countries_it = saves.keys(); countries_it.hasNext(); ) { saves_product_free = 0; saves_product_premium = 0; streams_sex_female_age_0_17 = 0; streams_sex_female_age_18_22 = 0; streams_sex_female_age_23_27 = 0; streams_sex_female_age_28_34 = 0; streams_sex_female_age_35_44 = 0; streams_sex_female_age_45_59 = 0; streams_sex_female_age_60_150 = 0; streams_sex_female_age_unknown = 0; streams_sex_male_age_0_17 = 0; streams_sex_male_age_18_22 = 0; streams_sex_male_age_23_27 = 0; streams_sex_male_age_28_34 = 0; streams_sex_male_age_35_44 = 0; streams_sex_male_age_45_59 = 0; streams_sex_male_age_60_150 = 0; streams_sex_male_age_unknown = 0; streams_sex_neutral_age_0_17 = 0; streams_sex_neutral_age_18_22 = 0; streams_sex_neutral_age_23_27 = 0; streams_sex_neutral_age_28_34 = 0; streams_sex_neutral_age_35_44 = 0; streams_sex_neutral_age_45_59 = 0; streams_sex_neutral_age_60_150 = 0; streams_sex_neutral_age_unknown = 0; streams_sex_unknown_age_0_17 = 0; streams_sex_unknown_age_18_22 = 0; streams_sex_unknown_age_23_27 = 0; streams_sex_unknown_age_28_34 = 0; streams_sex_unknown_age_35_44 = 0; streams_sex_unknown_age_45_59 = 0; streams_sex_unknown_age_60_150 = 0; streams_sex_unknown_age_unknown = 0; String country = countries_it.next(); streams_total = (int) ((JSONObject) streams.get(country)).get("total"); JSONObject sex = (JSONObject) ((JSONObject) streams.get(country)).get("sex"); for (Iterator sex_it = sex.keys(); sex_it.hasNext(); ){ String gender = sex_it.next(); if(gender.equals("female")) { JSONObject ages = (JSONObject) ((JSONObject) sex.get("female")).get("age"); for (Iterator age_it = ages.keys(); age_it.hasNext(); ) { String age = age_it.next(); if(age.equals("0-17")) streams_sex_female_age_0_17 += (int) ages.get("0-17"); else if(age.equals("18-22")) streams_sex_female_age_18_22 += (int) ages.get("18-22"); else if(age.equals("23-27")) streams_sex_female_age_23_27 += (int) ages.get("23-27"); else if(age.equals("28-34")) streams_sex_female_age_28_34 += (int) ages.get("28-34"); else if(age.equals("35-44")) streams_sex_female_age_35_44 += (int) ages.get("35-44"); else if(age.equals("45-59")) streams_sex_female_age_45_59 += (int) ages.get("45-59"); else if(age.equals("60-150")) streams_sex_female_age_60_150 += (int) ages.get("60-150"); else if(age.equals(age.equals("unknown"))) streams_sex_female_age_unknown += (int) ages.get("unknown"); } } else if(gender.equals("male")) { JSONObject ages = (JSONObject) ((JSONObject) sex.get("male")).get("age"); for (Iterator age_it = ages.keys(); age_it.hasNext(); ) { String age = age_it.next(); if(age.equals("0-17")) streams_sex_male_age_0_17 += (int) ages.get("0-17"); else if(age.equals("18-22")) streams_sex_male_age_18_22 += (int) ages.get("18-22"); else if(age.equals("23-27")) streams_sex_male_age_23_27 += (int) ages.get("23-27"); else if(age.equals("28-34")) streams_sex_male_age_28_34 += (int) ages.get("28-34"); else if(age.equals("35-44")) streams_sex_male_age_35_44 += (int) ages.get("35-44"); else if(age.equals("45-59")) streams_sex_male_age_45_59 += (int) ages.get("45-59"); else if(age.equals("60-150")) streams_sex_male_age_60_150 += (int) ages.get("60-150"); else if(age.equals(age.equals("unknown"))) streams_sex_male_age_unknown += (int) ages.get("unknown"); } } else if(gender.equals("neutral")) { JSONObject ages = (JSONObject) ((JSONObject) sex.get("neutral")).get("age"); for (Iterator age_it = ages.keys(); age_it.hasNext(); ) { String age = age_it.next(); if(age.equals("0-17")) streams_sex_neutral_age_0_17 += (int) ages.get("0-17"); else if(age.equals("18-22")) streams_sex_neutral_age_18_22 += (int) ages.get("18-22"); else if(age.equals("23-27")) streams_sex_neutral_age_23_27 += (int) ages.get("23-27"); else if(age.equals("28-34")) streams_sex_neutral_age_28_34 += (int) ages.get("28-34"); else if(age.equals("35-44")) streams_sex_neutral_age_35_44 += (int) ages.get("35-44"); else if(age.equals("45-59")) streams_sex_neutral_age_45_59 += (int) ages.get("45-59"); else if(age.equals("60-150")) streams_sex_neutral_age_60_150 += (int) ages.get("60-150"); else if(age.equals(age.equals("unknown"))) streams_sex_neutral_age_unknown += (int) ages.get("unknown"); } } else if(gender.equals("unknown")) { JSONObject ages = (JSONObject) ((JSONObject) sex.get("unknown")).get("age"); for (Iterator age_it = ages.keys(); age_it.hasNext(); ) { String age = age_it.next(); if(age.equals("0-17")) streams_sex_unknown_age_0_17 += (int) ages.get("0-17"); else if(age.equals("18-22")) streams_sex_unknown_age_18_22 += (int) ages.get("18-22"); else if(age.equals("23-27")) streams_sex_unknown_age_23_27 += (int) ages.get("23-27"); else if(age.equals("28-34")) streams_sex_unknown_age_28_34 += (int) ages.get("28-34"); else if(age.equals("35-44")) streams_sex_unknown_age_35_44 += (int) ages.get("35-44"); else if(age.equals("45-59")) streams_sex_unknown_age_45_59 += (int) ages.get("45-59"); else if(age.equals("60-150")) streams_sex_unknown_age_60_150 += (int) ages.get("60-150"); else if(age.equals(age.equals("unknown"))) streams_sex_unknown_age_unknown += (int) ages.get("unknown"); } } } saves_total = (int) ((JSONObject) saves.get(country)).get("total"); JSONObject products = (JSONObject) ((JSONObject) saves.get(country)).get("product"); for (Iterator product_it = products.keys(); product_it.hasNext(); ){ String product = product_it.next(); if(product.equals("premium")) saves_product_premium += (int) products.get("premium"); else if(product.equals("free")) saves_product_free += (int) products.get("free"); } TableRow row = new TableRow(); row.set("album_href", album_href); row.set("album_name", album_name); row.set("artists_hrefs", artists_hrefs); row.set("artists_names", artists_names); row.set("country", country); row.set("date", date); row.set("label", label); row.set("licensor", licensor); row.set("message_name", message_name); row.set("saves_product_free", saves_product_free); row.set("saves_product_premium", saves_product_premium); row.set("saves", saves_total); row.set("skips", skips); row.set("streams_sex_female_age_0_17", streams_sex_female_age_0_17); row.set("streams_sex_female_age_18_22", streams_sex_female_age_18_22); row.set("streams_sex_female_age_23_27", streams_sex_female_age_23_27); row.set("streams_sex_female_age_28_34", streams_sex_female_age_28_34); row.set("streams_sex_female_age_35_44", streams_sex_female_age_35_44); row.set("streams_sex_female_age_45_59", streams_sex_female_age_45_59); row.set("streams_sex_female_age_60_150", streams_sex_female_age_60_150); row.set("streams_sex_female_age_unknown", streams_sex_female_age_unknown); row.set("streams_sex_male_age_0_17", streams_sex_male_age_0_17); row.set("streams_sex_male_age_18_22", streams_sex_male_age_18_22); row.set("streams_sex_male_age_23_27", streams_sex_male_age_23_27); row.set("streams_sex_male_age_28_34", streams_sex_male_age_28_34); row.set("streams_sex_male_age_35_44", streams_sex_male_age_35_44); row.set("streams_sex_male_age_45_59", streams_sex_male_age_45_59); row.set("streams_sex_male_age_60_150", streams_sex_male_age_60_150); row.set("streams_sex_male_age_unknown", streams_sex_male_age_unknown); row.set("streams_sex_neutral_age_0_17", streams_sex_neutral_age_0_17); row.set("streams_sex_neutral_age_18_22", streams_sex_neutral_age_18_22); row.set("streams_sex_neutral_age_23_27", streams_sex_neutral_age_23_27); row.set("streams_sex_neutral_age_28_34", streams_sex_neutral_age_28_34); row.set("streams_sex_neutral_age_35_44", streams_sex_neutral_age_35_44); row.set("streams_sex_neutral_age_45_59", streams_sex_neutral_age_45_59); row.set("streams_sex_neutral_age_60_150", streams_sex_neutral_age_60_150); row.set("streams_sex_neutral_age_unknown", streams_sex_neutral_age_unknown); row.set("streams_sex_unknown_age_0_17", streams_sex_unknown_age_0_17); row.set("streams_sex_unknown_age_18_22", streams_sex_unknown_age_18_22); row.set("streams_sex_unknown_age_23_27", streams_sex_unknown_age_23_27); row.set("streams_sex_unknown_age_28_34", streams_sex_unknown_age_28_34); row.set("streams_sex_unknown_age_35_44", streams_sex_unknown_age_35_44); row.set("streams_sex_unknown_age_45_59", streams_sex_unknown_age_45_59); row.set("streams_sex_unknown_age_60_150", streams_sex_unknown_age_60_150); row.set("streams_sex_unknown_age_unknown", streams_sex_unknown_age_unknown); row.set("streams", streams_total); row.set("track_href", track_href); row.set("track_isrc", track_isrc); row.set("track_name", track_name); row.set("version", version); rows.add(row); out.output(row); } } } public static void main(String[] args) { Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class); Pipeline pipeline = Pipeline.create(options); PCollection input = pipeline.apply("Read Files", TextIO.read().from(options.getInputFile())); PCollection BQformat = input.apply(ParDo.of(new FormatBigQueryFn())); TableSchema schema = createBigQuerySchema(); BQformat.apply(BigQueryIO.writeTableRows() .to(options.getOutputDataset()) .withSchema(schema) .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND) .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)); pipeline.run(); } }