Atrix1987

private static final String UNKNOWN_VALUE = "unknown";
/**
 * Get Installer Id for the app
 *
 * @param context An instance of the application {@link Context}
 * @return The installer Id of the installer for the app
 */
public static String getInstallSource(Context context) {
  try {
    PackageManager pm = context.getPackageManager();

Atrix1987

#Get data
curl -u 649cxxx9663axxxx6042559e3:10yyy0c67zzz0 'https://amplitude.com/api/2/export?start=20161018T0&end=20161018T23' >> 20161018.zip
#Unzip The file
unzip 20161018.zip -d 20161018
#Remove folder nesting where 123456 is amplitude app id
mv 20161018/123456/*.gz 20161018
#Removing folders to maintain a decent structure
rmdir 20161018/123456
#Upload to GCS
gsutil -m cp -r 20161018 gs://BUCKET-NAME/sources/amplitude_/events/20161018

Atrix1987

//Configuring the options etc

Pipeline p = Pipeline.create(options);
p.apply(TextIO.Read.named("ReadFiles").from(options.getInputFile()).withCompressionType(CompressionType.GZIP))
  .apply(new ProcessRecords(eventType))
  .apply(BigQueryIO.Write.to(table.getTableReference()).withSchema(table.getSchema())
  .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
  .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

p.run();

Atrix1987

public static class ProcessRecords extends PTransform<PCollection<String>, PCollection<TableRow>> {

  private static final long serialVersionUID = -1916823923052893270L;

  private String eventName;

  public ProcessRecords(String eventName) {
    this.eventName = eventName;
  }

Atrix1987

private static class PrepareTableRow extends DoFn<String, TableRow> {

  private static final long serialVersionUID = -4178491975010026252L;

  private String eventName;

  public PrepareTableRow(String eventName) {
    this.eventName = eventName;
  }

Atrix1987

YD=`date --date="1 days ago" +%Y%m%d`
echo "Downloading files for $YD from amplitude for Sutori"
curl -u <API_KEY>:<api_secret> "https://amplitude.com/api/2/export?start=${YD}T0&end=${YD}T23" >> $YD.zip
echo 'Unzipping the downloaded files'
unzip "$YD.zip" -d "$YD"
echo 'Formatting folder structure'
mv "${YD}/<APP_ID>"/*.gz "$YD"
rmdir "${YD}"/<APP_ID>
echo 'Uploading to GCS'
gsutil -m cp -r "${YD}" gs://<bucket_name>/sources/amplitude_/events/"${YD}"

Atrix1987

SELECT
  integer(mnthid/100) YEAR,
  integer(mnthid%100) MONTH,
  NewUsers,
  Recurring,
  Resurrected,
  NewUsers+Recurring+Resurrected TOTAL FROM (
  SELECT
    mnthid,
    NewUsers,

Atrix1987

SELECT  YR, month, NewUsers, Recurring, Resurrected, NewUsers+Recurring+Resurrected TOTAL FROM (
  SELECT month, YR, SUM(IF(DIFF IS NULL, uc, 0)) NewUsers, SUM(IF(DIFF = 1, uc, 0)) Recurring, SUM(IF(DIFF > 1, uc, 0)) Resurrected FROM (
    SELECT month, DIFF, YR, EXACT_COUNT_DISTINCT(amplitude_id) uc FROM (
      SELECT amplitude_id, month, YR, month-prevMonth AS DIFF FROM (
        SELECT amplitude_id, month, YR, LAG(month,1) OVER (PARTITION BY amplitude_id ORDER BY month ) AS prevMonth FROM (
          SELECT amplitude_id, YEAR(server_upload_time) YR, MONTH(server_upload_time) AS month FROM (
            SELECT amplitude_id, server_upload_time FROM
            	[gproject:dataset_name.amplitude_start_session])
     		GROUP BY 1, 2, 3
     	)

Atrix1987

{
  "Comment": "Load Data from Firebase and send it to AWS Cloud Search",
  "StartAt": "CalculateInterval",
  "States": {
    "Get data from Firebase": {
      "Type": "Parallel",
      "Next": "Format Parallel Processed Data",
      "Branches": [
        {
          "StartAt": "Fetch User Data",

Atrix1987

curl -u API_Key:Secret_Key 'https://amplitude.com/api/2/export?start=20150201T5&end=20150203T20' >> yourfilename