Java sample to backfill meters from MongoDB

java sample to backfill meters from historical data stored in mongodb into amberflo the sample will perform the following tasks read entries from mongodb collection in this sample, events are ingested as meters into amberflo for each entry, ingest an event meter into amberflo ensure the create time is used as ingested event time in utc use the entry id as uniqueid of the ingested meter to prevent duplicated ingested records if the code were to run multiple times maven dependencies (check for latest versions) \<dependencies> \<dependency> \<groupid>io amberflo\</groupid> \<artifactid>metering java client\</artifactid> \<version>1 3 6\</version> \</dependency> \<dependency> \<groupid>org mongodb\</groupid> \<artifactid>mongo java driver\</artifactid> \<version>3 5 0\</version> \</dependency> \<dependency> \<groupid>com stripe\</groupid> \<artifactid>stripe java\</artifactid> \<version>20 52 0\</version> \</dependency> \</dependencies> package com amberflo backfill; import com amberflo metering ingest meteringcontext; import com amberflo metering ingest meter message domain; import com amberflo metering ingest meter message metermessage; import com amberflo metering ingest meter message metermessagebuilder; import com amberflo metering ingest meter message region; import com mongodb mongoclient; import com mongodb mongoclienturi; import com mongodb client finditerable; import com mongodb client mongocollection; import com mongodb client mongocursor; import com mongodb client mongodatabase; import org bson document; import java time instant; import java time localdatetime; import java time zoneid; import java util hashmap; import java util map; import static com amberflo metering ingest meteringcontext metering; public class backfillmongodbdata { //todo set the database connection, queries, and amberflo info private static final string mongo host = "sample db mongo host"; private static final int mongo port = 27017; private static final string mongo user = "sample user"; private static final string mongo pass = "pass"; private static final string db name = "sampledb"; private final static string amberflo api key = "amberflo api key"; //end todo private final static string event create = "event create"; public static void main(final string\[] args) throws exception { meteringcontext createorreplacecontext( amberflo api key, "events api", domain prod, region us west, 1, 10 ); // mongodb connection string final string connectionstring = string format("mongodb //%s %s@%s %s", mongo user, mongo pass, mongo host, mongo port); final mongoclienturi uri = new mongoclienturi(connectionstring); // connecting to the mongodb server using the given client uri final mongoclient mongoclient = new mongoclient(uri); // fetching the database from the mongodb final mongodatabase db = mongoclient getdatabase(db name); // fetching the collection from the mongodb final mongocollection\<document> col = db getcollection("events"); // performing a read operation on the collection final finditerable\<document> fi = col find(); final mongocursor\<document> cursor = fi iterator(); int counter = 1; try { while (cursor hasnext()) { final document document = cursor next(); if (document == null) { continue; } //example of a nested document final document eventdata = (document) document get("eventdata"); final string eventsourcetype = eventdata get("event source type") tostring(); final string eventtype = eventdata get("event type") tostring(); final string metername = event create; final string customerid = eventdata get("customer id") tostring(); final string id = document get(" id") tostring(); //converting timestamp stored as double epoch final long eventtimeepoch = ((double) double parsedouble(document get("created time epoch utc") tostring())) longvalue(); final localdatetime eventtime = instant ofepochmilli(eventtimeepoch) atzone(zoneid of("utc")) tolocaldatetime(); system out println(string format("ingesting event source type=%s, customerid=%s, id=%s, eventtime=%s, eventtype=%s, metername %s", eventsourcetype, customerid, id, eventtime, eventtype, metername)); system out println(document tojson()); final map\<string, string> dimensions = new hashmap(); dimensions put("event source type", eventsourcetype); dimensions put("event type", eventtype); final metermessage metermessage = metermessagebuilder createinstance(metername, eventtime, customerid) setmetervalue(1) setuniqueid(id) setdimensionsmap(dimensions) build(); metering() meter(metermessage); if (counter % 100 == 0) { thread sleep(5000); } counter++; } system out println("total meters ingested " + counter); } finally { cursor close(); mongoclient close(); meteringcontext flushandclose(); system out println("done!"); } } }