Forem: Peter Milne

Near Real-Time Campaign Reporting Part 3 - Campaign Service and Campaign UI

Peter Milne — Sat, 29 Feb 2020 17:26:46 +0000

This is the third in a series of articles describing a simplified example of near real-time Ad Campaign reporting on a fixed set of campaign dimensions usually displayed for analysis in a user interface. The solution presented in this series relies on Kafka, Aerospike’s edge-to-core data pipeline technology, and Apollo GraphQL

Part 1: real-time capture of Ad events via Aerospike edge datastore and Kafka messaging.
Part 2: aggregation and reduction of Ad events via Aerospike Complex Data Type (CDT) operations into actionable Ad Campaign Key Performance Indicators (KPIs).
Part 3: displaying the KPIs in an Ad Campaign user interface displays those KPIs using GraphQL to retrieve data stored in an Aerospike Cluster.

Data flow

Summary of Part 1 and Part 2

In part 1, we

used an ad event simulator for data creation
captured that data in the Aerospike “edge” database
pushed the results to a Kafka cluster via Aerospike’s Kafka Connector

In part 2, we then

consumed events from Kafka exported via Aerospike’s Kafka Connector
aggregated each event into Campaign KPIs on arrival
published a message in Kafka containing the new KPI value

Parts 1 and 2 form the base for Part 3

The use case — Part 3

Part 3 has two use cases:

displaying Campaign details in a UI
updating Campaign KPIs in real-time

As mentioned in Part 2, the KPIs in this example are very simple counters, but the same techniques could be applied to more sophisticated measurements such as histograms, moving averages and trends.

The first use case reads the Campaign details, including the KPIs from Aerospike record.

The second use case is based on a GraphQL subscription specific to a Campaign and KPI. A subscription message with the new KPI value is sent from the campaign-service to the campaign-ui when the KPI has changed.

To recap - the Aerospike record looks like this:

Bin	Type	Example value
c-id	long	6
c-date	long	1579373062016
c-name	string	Acme campaign 6
stats	map	{"visits":6, "impressions":78, "clicks":12, "conversions":3}

The Core Aerospike cluster is configured to prioritise consistency over availability to ensure that numbers are accurate and consistent.

This sequence diagram shows the use cases:

On page load
KPI update

Campaign Service and UI scenarios

Companion code

The companion code is in GitHub. The complete solution is in the master branch. The code for this article is in the ‘part-3’ branch.

Javascript and Node.js are used in each back-end services, although the same solution is possible in any language.

The solution consists of:

All of the service and containers in Part 1 and Part 2.
Campaign service - Node.js and Apollo GraphQL Server
Campaign UI - React, Material UI and Apollo GraphQL Client React

Docker and Docker Compose simplify the setup to allow you to focus on the Aerospike specific code and configuration.

What you need for the setup

All the prerequisites are described in Part 1.

Setup steps

To set up the solution, follow these steps. The Docker images are built by downloading resources, be aware that the time to download and build the software depends on your internet bandwidth and your computer.

Follow the setup steps in Part 1. Then

Step 1. Checkout the part-3 branch

$ git checkout part-3

Step 2. Then run the following command to delete the Aerospike data and the Kafka topics data.

$ ./delete-data.sh

Step 3. Finally run

$ docker-compose up -d
$ docker-compose logs -f publisher-simulator

Once up and running, after the services have stabilised, you will see the output in the console similar to this:

Sample console output

Step 4. Go to the UI with this URL:

http://localhost:5000/

to display the Campaign application

Campaign KPI application

Note: you are now running 12 services on your local machine.

How do the components interact?

Component Interaction

Docker Compose orchestrates the creation of twelve services in separate containers:

All of the services and containers of Part 1 and Part 2 with the addition of:

Campaign Service campaign-service - A node.js and Apollo GraphQL Server service

Like the services in Part 1 and Part 2, the campaign-service uses the Aerospike Node.js client. On the first build, all the service containers that use Aerospike will download and compile the supporting C library.

As mentioned in Part 1 and Part 2, the Dockerfile for each container uses multi-stage builds to minimise the number of times the C library is compiled.

Campaign UI campaign-ui - A React and Material UI single-page web application to display Campaign KPIs, it uses the Apollo Client React GraphQL client.

How is the solution deployed?

Each container is deployed using docker-compose on your local machine.

Note: The campaign-service and campaign-ui containers is deployed along with all the containers from Part 1 and Part 2.

Deployment

How does the solution work?

Campaign service

The campaign-service is a deliberately simple Apollo Server providing a GraphQL schema and the resolvers for the root operations defined in that schema.

index.js

src/index.js contains:

a GraphQL server
a schema in Schema Definition Language
resolvers for the root operations

Note: this is an example server only and is not structured for production.

Schema definition

The schema defines the types of:

Campaign - Campain meta data
CampaignKPI - the set of KPIs for a Campaign
KPI - a single KPI e.g. impressions

Queries of:

campaign(id:ID!) - returning a single Campaign
campaigns(ids:[ID!]!) - returning a set of campaigns matching the Ids passed

and Subscriptions of:

kpiUpdate(campaignId:ID!, kpiName:String) - posts a KPI event when a KPI update occurs matching the campaignId and kpiName

  type Campaign {
    id: ID
    name: String
    aggregateKPIs: CampaignKPI
  }

  type CampaignKPI {
    clicks: Int
    impressions: Int
    visits: Int
    conversions: Int
  }

  type KPI {
    campaignId: ID
    name: String
    value: Int
  }

  type Query {
    campaign(id:ID):Campaign
    campaigns(ids: [ID!]!): [Campaign]
  }

  type Subscription {
    kpiUpdate(campaignId:ID!, kpiName:String):KPI
  }

GraphQL schema

Resolvers

Each field in GraphQL can have a resolver function defined to resolve the value of the field.

In this schema we have defined resolvers for:

Query
- campaign(...)
- campaigns(...)
Subscription
- kpiUpdate(...)

Query resolver function names match the field names of campaign and campaigns and they delegate to the campaign data source CampaignDataSource.js.

  Query: {
    campaign: (_1, args, context, _2) => {
      return context.campaignsDS.fetchCampaign(args.id);
    },

    campaigns: (_1, args, context, _3) => {
      return context.campaignsDS.fetchCampaignsById(args.ids);
    }
  },

Query resolvers

The single Subscription resolver kpiUpdate implements a filter allowing the front end to subscribe to a KPI for a specific campaign and KPI name.

  Subscription: {
    kpiUpdate: {
      subscribe: withFilter(
        (parent, args, context, info) => pubsub.asyncIterator(['NEW_KPI']),
        (payload, variables) => {
          let isFiltered = (variables.campaignId == payload.campaignId.toString() &&
            variables.kpiName == payload.kpi);
          if (isFiltered)
            console.log(`Subscribe: payload ${JSON.stringify(payload)}, variables ${JSON.stringify(variables)}`);
          return isFiltered;
        }),
      resolve: (payload) => {
        let event = {
          campaignId: payload.campaignId,
          name: payload.kpi,
          value: payload.value
        };
        console.log(`kpiUpdate:`, event);
        return event;
      },
    },
  }

Subscription resolver

It is a surprisingly small amount of code to implement a GraphQl Schema and Server.

CampaignDataSource.js

src/CampaignDataSource.js is the connector to Aerospike, whose job is to read aerospike Campaign records and transform them to the type described in the GraphQL schema.

Fetching a single record by ID

Fetching a single campaign is implemented using the Aerospike get operation. The whole Aerospike record is read using the primary key and transformed into the GraphQL type. (see Transforming a record to a Campaign)

  async fetchCampaign(id) {
    try {
      let client = await asClient();
      let key = new Aerospike.Key(config.namespace, config.campaignSet, parseInt(id));
      let record = await client.get(key);
      return campaignFromRecord(record);
    } catch (err) {
      if (err.code && err.code == 2) {
        throw new ApolloError(`Campaign ${id} not found`);
      } else {
        console.error('Fetch campaign error:', err);
        throw new ApolloError(`Fetch campaign by ID: ${id}`, err);
      }
    }
  }

Fetching multiple records an array of IDs

To fetch multiple Campaign records we use the Aerospike batchRead operation. The batchRead operation reads the requested records concurrently, this is very efficient in a multi-node cluster as records are evenly distributed across nodes and each node will do about the same amount of work to locate and return the requested records.

  async fetchCampaignsById(campaignIds) {
    try {
      let client = await asClient();
      let keys = campaignIds.map((id) => {
        return {
          key: new Aerospike.Key(config.namespace, config.campaignSet, parseInt(id)),
          read_all_bins: true
        };
      });
      let records = await client.batchRead(keys);
      records = records.filter(n => n.status == 0);
      let campaigns = records.map((element) => {
        return campaignFromRecord(element.record);
      });
      return campaigns;
    } catch (err) {
      console.error(`fetchCampaignsById: ${campaignIds}`, err);
      throw new ApolloError(`fetchCampaignsById: ${campaignIds}`, err);
    }
  }

Fetching multiple records using a query

This function is not actually used in the solution, but it does illustrate how to use Aerospike's query capability based on a secondary index and filters.

  async listCampaigns() {
    try {
      let campaigns = [];

      let client = await asClient();
      let query = client.query(config.namespace, config.campaignSet);

      // filter by campaign date for today -- demo only
      let startDate = new Date();
      startDate.setHours(0);
      startDate.setMinutes(0);
      startDate.setSeconds(0);
      startDate.setMilliseconds(0);
      let endDate = new Date(startDate);
      endDate.setHours(23);
      endDate.setMinutes(59);
      endDate.setSeconds(59);
      endDate.setMilliseconds(999);

      query.where(Aerospike.filter.range(config.campaignDate, startDate.getTime(), endDate.getTime()));

      let stream = query.foreach();

      return new Promise((resolve, reject) => {
        stream.on('data', (record) => {
          let campaign = campaignFromRecord(record);
          campaigns.push(campaign);
        });
        stream.on('error', (error) => {
          console.error('Aerospike select error', error);
          reject(error);
        });
        stream.on('end', () => {
          resolve(campaigns);
        });
      });
    } catch (err) {
      console.error(`List campaigns error:`, err);
      throw new ApolloError(`List campaigns error:`, err);
    }
  }

Transforming a record to a Campaign

A Campaign record is stored in a set of Bins, and these need to be transformed to the GraphQL type.

Aerospike record	GraphQL types
{ "c-id": 10, "stats": { "visits": 0, "impressions": 0, "clicks": 0, "conversions": 0 }, "c-name": "Acme campaign 10", "c-date": 1581683864910 }	type Campaign { id: ID name: String aggregateKPIs: CampaignKPI } type CampaignKPI { clicks: Int impressions: Int visits: Int conversions: Int }

Aerospike record

GraphQL types

  {
    "c-id": 10,
    "stats": {
      "visits": 0,
      "impressions": 0,
      "clicks": 0,
      "conversions": 0
    },
    "c-name": "Acme campaign 10",
    "c-date": 1581683864910
  }

  type Campaign {
    id: ID
    name: String
    aggregateKPIs: CampaignKPI
  }
  type CampaignKPI {
    clicks: Int
    impressions: Int
    visits: Int
    conversions: Int
  }

The function takes the Aerosike record and returns a Campaign type:

const campaignFromRecord = (record) => {
  let campaign = {
    id: record.bins[config.campaignIdBin],
    name: record.bins[config.campaignNameBin],
    aggregateKPIs: record.bins[config.statsBin]
  };
  return campaign;
};

KpiReceiver

The KpiReceiver listens to the Kafka topic subscription-events and when a message is received, it is published as a GraphQL subscription. Using Kafka as the pubsub technology allows the campaign-service to scale without the KPI event being lost.

Most of the work is done in this code:

    this.consumer.on('message', async function (eventMessage) {
      try {
        let payload = JSON.parse(eventMessage.value);
        pubsub.publish('NEW_KPI', payload);
      } catch (error) {
        console.error(error);
      }
    });

Note: pubsub (line 4) as part of the apollo-server npm package and does all the heavy lifting in implementing GraphQL subscriptions. The pubsub
reference is passed into the constructor:

 constructor(pubsub) {
    ...
    this.pubsub = pubsub;
    ...
  }

Campaign UI

The campaign-ui is a single page web application implemented using React, Material UI and Apollo GraphQL Client React.

The application is implemented by composing the Components:

ApolloProvider
- App
  - CampaignList
    - CampaignRow
      - Kpi

index.js

Setting up a React application to use Apollo GraphQL is quite straight forward by following this guide.

In our code we will use GraphQL Subscriptions implemented with websockets and Apollo provides all the helper classes and functions to achieve this.

First we create a link to our GraphQL server:

const httpLink = new HttpLink({
  uri: `http://${campaignServiceHost}:${campaignServicePort}`,
});

then we create a web socket link:

const wsLink = new WebSocketLink({
  uri: `ws://${campaignServiceHost}:${campaignServiceWsPort}/graphql`,
  options: {
    reconnect: true,
    lazy: true,
  },
});

We can optimise the communications paths to the server by splitting the link based on the operation type.

const link = split(
  // split based on operation type
  ({ query }) => {
    const definition = getMainDefinition(query);
    return (
      definition.kind === 'OperationDefinition' &&
      definition.operation === 'subscription'
    );
  },
  wsLink,
  httpLink,
);

We also add a client side cache - not necessary in this example, but fun to add anyway.

const cache = new InMemoryCache({
  dataIdFromObject: defaultDataIdFromObject,
});

Finally we create an ApolloClient instance

const client = new ApolloClient({
  link,
  cache
});

ApolloProvider is a HOC from Apollo that encapsulates the App component and passes down the ApolloClient instance as a property of the ApolloProvider and this client is available to child components of App.

const WrappedApp = (
  <ApolloProvider client={client}>
    <App />
  </ApolloProvider>
);

The React App is ready to interact with the campaign-service.

CampaignList

CampaignList.js is a table using Material-UI components. An array of campaign Ids is passed in props. These Ids are used in the GraphQL query:

const CAMPAIGN_LIST = gql`
query campaigns($campaignIds: [ID!]!) {
  campaigns(ids: $campaignIds) {
    id
    name
    aggregateKPIs {
      clicks
      impressions
      visits
      conversions
    }
  }
}
`;

Campaign query

The render() method creates a TableContainer with a TableHeader, each row in the table is CampaignRow component.

  return (
    <TableContainer component={Paper}>
      <Table className={classes.table} size="small" aria-label="dense table">
        <TableHead>
          <TableRow>
            <TableCell className={classes.kpiColumn} >Id</TableCell>
            <TableCell className={classes.campaignColumn}>Campaign Name</TableCell>
            <TableCell className={classes.kpiColumn} align="right">Impressions</TableCell>
            <TableCell className={classes.kpiColumn} align="right">Clicks</TableCell>
            <TableCell className={classes.kpiColumn} align="right">Visits</TableCell>
            <TableCell className={classes.kpiColumn} align="right">Conversions</TableCell>
          </TableRow>
        </TableHead>
        <TableBody>
          {campaignList}
        </TableBody>
      </Table>
    </TableContainer >
  );

render() method

CampaignRow

The CamaignRow component receives the "campaign" via props. Each KPI column is implemented using the Kpi component.

export default function CampaignRow({ campaign }) {
  return (
    <TableRow key={campaign.id}>
      <TableCell component="th" scope="row">{campaign.id}</TableCell>
      <TableCell align="left" >{campaign.name}</TableCell>
      <TableCell align="right"><Kpi campaignId={campaign.id} kpiName="impressions" initialValue={campaign.aggregateKPIs.impressions} /></TableCell>
      <TableCell align="right"><Kpi campaignId={campaign.id} kpiName="clicks" initialValue={campaign.aggregateKPIs.clicks} /></TableCell>
      <TableCell align="right"><Kpi campaignId={campaign.id} kpiName="visits" initialValue={campaign.aggregateKPIs.visits} /></TableCell>
      <TableCell align="right"><Kpi campaignId={campaign.id} kpiName="conversions" initialValue={campaign.aggregateKPIs.conversions} /></TableCell>
    </TableRow>
  )
}

CampaignRow component

Kpi

The Kpi component renders the KPI value and, more interestingly, subscribes to the defigned GraphQL subscription kpiUpdate.

const KPI_SUBSCRIPTION = gql`
subscription kpiUpdate($campaignId: ID!, $kpiName:String!){
  kpiUpdate(campaignId: $campaignId, kpiName: $kpiName) {
    campaignId
    name
    value
  }
}
`;

GraphQL Subscription

The component is rendered by including the GraphQL subscription to listen for KPI updates.

  render() {
    const { startAttention } = this.state
    const variant = startAttention ? 'H5' : 'inherit';
    const type = startAttention ? 'secondary' : 'inherit';
    return (
      <Typography color={type} variant={variant}>
        <Subscription subscription={KPI_SUBSCRIPTION}
          variables={{ campaignId: this.state.campaignId, kpiName: this.state.kpiName }}
          shouldResubscribe={true} onSubscriptionData={this.attention}>
          {
            ({ data, loading }) => {
              if (data) {
                return (data.kpiUpdate.value);
              }
              return (this.state.initialValue);
            }
          }
        </Subscription >
      </Typography>
    );
  }

render() method

In order to highlight the change in a KPI value, the new value is turned red for about 1 second.

  attention(something) {
    this.setState({ startAttention: true })
    setTimeout(() => this.setState({ startAttention: false }), 1000);
  }

The whole story

In this series we have used Aerospike and Kafka to build a simple edge-to-core solution to capture real-time ad events for Campaign Reporting.

The "Edge" portion of the solution would be deployed geographically and the "Core" would be deploy centrally using a hub and spoke pattern.

Geographical deployment

Event sequence

Users interact with ads on Publisher websites and the interaction events are sent to the local "edge" event collector and datastore.

Event data is propagated to the "core" aggregator/reducer using Kafka. The aggregator/reducer takes each event and aggregates it with the designated KPI, in this example, the KPIs are simple atomic counters stored in an Aerospike CDT.

Using the Campaign UI, campaign specialists can monitor the KPIs in real-time for campaign optimisation. KPIs are updated live, without costly page loads or polling, using GraphQL subscriptions.

Event sequence

The complete component interaction

The complete component diagram shows all the components, their packages and their interaction with each other. The packages are dockerized for deployment, vastly reducing the risk of production failures due to a missing dependency. We have used docker-compose in this example; in a real-world environment, Kubernetes is a great choice for container orchestration.

Review

Part 1 of this series describes:

creating mock Campaign data
a publisher simulator
an event receiver
an edge database
an edge exporter

Part 2 describes the aggregation and reduction of Ad events into Campaign KPIs using Kafka as the messaging system and Aerospike as the consistent data store.

Part 3 (this article) describes the Campaign service and Campaign UI to for a user to view the Campaign KPIs in near real-time.

Possible improvements to the entire application

The KPI data structure (cube) and event aggregation are deliberately simple to illustrate the technologies used. Here are a few improvement ideas:
- Event aggregation using Spark. Spark streaming can be used for more sophisticated aggregation and reduction. Aerospike provides a Spark Data Frame which makes integration of Spark and Aerospike easy.
- The Campaign Service and UI. These are very basic and show only a portion of the possible functionality. Campaign management is complex and a complete campaign management service and UI are outside the scope of this article. A simple improvement would provide functionality for the user to specify the campaigns to monitor.

Disclaimer

This article, the code samples, and the example solution are entirely my own work and they are not endorsed by Aerospike, Confluent or Apollo. The code is PoC quality only and it is not production strength and is available to anyone under the MIT License.

Near real-time Campaign Reporting Part 2 - Aggregation/Reduction

Peter Milne — Tue, 04 Feb 2020 12:09:47 +0000

This is the second in a series of articles describing a simplified example of near real-time Ad Campaign reporting on a fixed set of campaign dimensions usually displayed for analysis in a user interface. The solution presented in this series relies on Kafka, Aerospike’s edge-to-core data pipeline technology, and Apollo GraphQL

Part 1: real-time capture of Ad events via Aerospike edge datastore and Kafka messaging.
Part 2: aggregation and reduction of Ad events leveraging Aerospike Complex Data Types (CDTs) for aggregation and reduction of Ad events into actionable Ad Campaign Key Performance Indicators (KPIs).
Part 3: describes how an Ad Campaign user interface displays those KPIs using GraphQL retrieve data stored in an Aerospike Cluster.

Data flow

Summary of Part 1

In part 1 of this series, we

used an ad event simulator for data creation
captured that data in the Aerospike “edge” database
pushed the results to a Kafka cluster via Aerospike’s Kafka Connector

Part 1 is the base used to implement Part 2

The use case — Part 2

The simplified use case for Part 2 consists of reading Ad events from a Kafka topic, aggregating/reducing the events into KPI values. In this case the KPIs are simple counters, but in the real-world these would be more complex metrics like averages, gauges, histograms, etc.

The values are stored in a data cube implemented as a Document or Complex Data Type (CDT) in Aerospike. Aerospike provides fine-grained operations to read or write one or more parts of a CDT in a single, atomic, database transaction.

The Aerospike record:

Bin	Type	Example value
c-id	long	6
c-date	long	1579373062016
c-name	string	Acme campaign 6
stats	map	{"visits":6, "impressions":78, "clicks":12, "conversions":3}

The Core Aerospike cluster is configured to prioritise consistency over availability to ensure that numbers are accurate and consistent for use with payments and billing. Or in others words: Money

In addition to aggregating data, the new value of the KPI is sent via another Kafka topic (and possible separate Kafka cluster) to be consumed by the Campaign Service as a GraphQL subscription and providing a live update in the UI. Part 3 covers the Campaign Service, Campaign UI and GraphQL in detail.

Aggregation/Reduction sequence

Companion code

The companion code is in GitHub. The complete solution is in the master branch. The code for this article is in the part-2 branch.

Javascript and Node.js are used in each service although the same solution is possible in any language.

The solution consists of:

All of the service and containers in Part 1.
Aggregator/Reducer service - Node.js

Docker and Docker Compose simplify the setup to allow you to focus on the Aerospike specific code and configuration.

What you need for the setup

All the perquisites are described in Part 1.

Setup steps

To set up the solution, follow these steps. Because executable images are built by downloading resources, be aware that the time to download and build the software depends on your internet bandwidth and your computer.

Follow the setup steps in Part 1. Then

Step 1. Checkout the part-2 branch

$ git checkout part-2

Step 2. Then run

$ docker-compose up

Once up and running, after the services have stabilised, you will see the output in the console similar to this:

Sample console output

How do the components interact?

Component Interaction

Docker Compose orchestrates the creation of several services in separate containers:

All of the services and containers in Part 1 with the addition of:

Aggregator/Reducer aggregator-reducer - A node.js service to consume Ad event messages from the Kafka topic edge-to-core and aggregates the single event with the existing data cube. The data cube is a document stored in an Aerospike CDT. A CDT document can be a list, map, geospatial, or nested list-map in any combination. One or more portions of a CDT can be mutated and read in a single atomic operation. See CDT Sub-Context Evaluation

Here we use a simple map where multiple discrete counters are incremented. In a real-world scenario, the datacube would be a complex document denormalized for read optimization.

Like the Event Collector and the Publisher Simulator, the Aggregator/Reducer uses the Aerospike Node.js client. On the first build, all the service containers that use Aerospike will download and compile the supporting C library. The Dockerfile for each container uses multi-stage builds to minimise the number of times the C library is compiled.

Kafka Cli kafkacli - Displays the KPI events used by GraphQL in Part 3.

How is the solution deployed?

Each container is deployed using docker-compose on your local machine.

Note: The aggregator-reducer container is deployed along with all the containers from Part 1.

Deployment

How does the solution work?

The aggregator-reducer is a headless service that reads a message from the Kafka topic edge-to-core. The message is the whole Aerospike record written to edge-aerospikedb and exported by edge-exporter.

The event data is extracted from the message and written to core-aerospikedb using multiple CDT operations in one atomic database operation.

aggregation flow

Connecting to Kafka

To read from a Kafka topic you need a Consumer and this is configured to read from one or more topics and partitions. In this example, we are reading a message from one topic edge-to-core and this topic has only 1 partition.

    this.topic = {
      topic: eventTopic,
      partition: 0
    };
    this.consumer = new Consumer(
      kafkaClient,
      [],
      {
        autoCommit: true,
        fromOffset: false
      }
    );

    let subscriptionPublisher = new SubscriptionEventPublisher(kafkaClient);

    addTopic(this.consumer, this.topic);

    this.consumer.on('message', async function (eventMessage) {
    ...
    });

    this.consumer.on('error', function (err) {
      ...
    });


    this.consumer.on('offsetOutOfRange', function (err) {
      ...
    });

Note that the addTopic() is called after the Consumer creation. This function attempts to add a topic to the consumer, if unsuccessful it waits 5 seconds and tries again. Why do this? The Consumer will throw an error if the topic is empty and this code overcomes that problem.

const addTopic = function (consumer, topic) {
  consumer.addTopics([topic], function (error, thing) {
    if (error) {
      console.error('Add topic error - retry in 5 sec', error.message);
      setTimeout(
        addTopic,
        5000, consumer, topic);
    }
  });
};

Extract the event data

The payload of the message is a complete Aerospike record serialised as JSON.

{
  "msg": "write",
  "key": [
    "test",
    "events",
    "AvYAdzrOmm1xwvWZFyGwEvgjwnk=",
    null
  ],
  "gen": 0,
  "exp": 0,
  "lut": 0,
  "bins": [
    {
      "name": "event-id",
      "type": "str",
      "value": "0d33124c-beca-4b4f-a833-8c9646167e8c"
    },
    {
      "name": "event-data",
      "type": "map",
      "value": {
        "geo": [
          52.366521,
          4.894981
        ],
        "tag": "0a3ca7c5-b845-49ed-ab3a-129f9eca23d6",
        "publisher": "e5b08db3-07b5-456b-aaac-1e59f76c4dd6",
        "event": "impression",
        "userAgent": "Mozilla/5.0 (X11; CrOS x86_64 8172.45.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.64 Safari/537.36"
      }
    },
    {
      "name": "event-tag-id",
      "type": "str",
      "value": "0a3ca7c5-b845-49ed-ab3a-129f9eca23d6"
    },
    {
      "name": "event-source",
      "type": "str",
      "value": "e5b08db3-07b5-456b-aaac-1e59f76c4dd6"
    },
    {
      "name": "event-type",
      "type": "str",
      "value": "impression"
    }
  ]
}

These items are extracted:

Event value
Tag id
Event source

These values are used in the aggregation step.

let payload = JSON.parse(eventMessage.value);
// Morph the array of bins to and object
let bins = payload.bins.reduce(
  (acc, item) => {
    acc[item.name] = item;
    return acc;
  },
  {}
);
// extract the event data value
let eventValue = bins['event-data'].value;
// extract the Tag id
let tagId = eventValue.tag;
// extract source e.g. publisher, vendor, advertiser
let source = bins['event-source'].value;

Lookup Campaign Id using Tag

The Tag Id is used to locate the matching Campaign. During campaign creation, a mapping between Tags and Campaign is created, this example uses an Aerospike record where the key is the Tag id and the value is the Campaign Id, and in this case, Aerospike is used a Dictionary/Map/Associative Array.

//lookup the Tag id in Aerospike to obtain the Campaign id
let tagKey = new Aerospike.Key(config.namespace, config.tagSet, tagId);
let tagRecord = await aerospikeClient.select(tagKey, [config.campaignIdBin]);
// get the campaign id
let campaignId = tagRecord.bins[config.campaignIdBin];

Aggregating the Event

The Ad event is specific to a Tag and therefore a Campaign. In our model, a Tag is directly related to a Campaign and KPIs are collected at the Campaign level. In the real-world KPIs are more sophisticated and campaigns have many execution plans (line items).

Each event for a KPI increments the value by 1. Our example stores the KPIs in a document structure (CDT) in a bin in the Campaign record. Aerospike provides operations to atomically access and/or mutate sub-contexts of this structure to ensure the operation latency is ~1ms.

In a real-world scenario, events would be aggregated with sophisticated algorithms and patterns such as time-series, time windows, histograms, etc.

Our code simple increments the value KPI value by 1 using the KPI name as the 'path' to the value:

const accumulateInCampaign = async (campaignId, eventSource, eventData, asClient) => {
  try {
    // Aerospike CDT operation returning the new DataCube
    let campaignKey = new Aerospike.Key(config.namespace, config.campaignSet, campaignId);
    const kvops = Aerospike.operations;
    const maps = Aerospike.maps;
    const kpiKey = eventData.event + 's';
    const ops = [
      kvops.read(config.statsBin),
      maps.increment(config.statsBin, kpiKey, 1),
    ];
    let record = await asClient.operate(campaignKey, ops);
    let kpis = record.bins[config.statsBin];
    console.log(`Campaign ${campaignId} KPI ${kpiKey} processed with result:`, JSON.stringify(record.bins, null, 2));
    return {
      key: kpiKey,
      value: kpis
    };
  } catch (err) {
    console.error('accumulateInCampain Error:', err);
    throw err;
  }
};

The new KPI value is incremented and the new value is returned. The magic of Aerospike ensures that the operation is Atomic and Consistent across the cluster with a latency of about 1 ms.

Publishing the new KPI

We could stop here and allow the Campaign UI and Service (Part 3) to poll the Campaign store core-aerospikedb to obtain the latest campaign KPIs - this is a typical pattern.

A more advanced approach is to stimulate the UI whenever a value has changed or at a specified frequency. While introducing new technology and challenges, this approach offers a very responsive UI presenting up to the second KPI values to the user.

The SubScriptionEventPublisher uses Kafka as Pub-Sub to publish the new KPI value for a specific campaign on the topic subscription-events. In Part 3 the campaign-service receives this event and publishes it as a GraphQL Subscription

class SubscriptionEventPublisher {
  constructor(kafkaClient) {
    this.producer = new HighLevelProducer(kafkaClient);
  };

  publishKPI(campaignId, accumulatedKpi) {
    const subscriptionMessage = {
      campaignId: campaignId,
      kpi: accumulatedKpi.key,
      value: accumulatedKpi.value
    };
    const producerRequest = {
      topic: subscriptionTopic,
      messages: JSON.stringify(subscriptionMessage),
      timestamp: Date.now()
    };

    this.producer.send([producerRequest], function (err, data) {
      if (err)
        console.error('publishKPI error', err);
      // else
      // console.log('Campaign KPI published:', subscriptionMessage);
    });
  };
}

Review

Part 1 of this series describes:

creating mock Campaign data
a publisher simulator
an event receiver
an edge database
an edge exporter

This article (Part 2) describes the aggregation and reduction of Ad events into Campaign KPIs using Kafka as the messaging system and Aerospike as the consistent data store.

Part 3 describes the Campaign service and Campaign UI to for a user to view the Campaign KPIs in near real-time.

Disclaimer

This article, the code samples, and the example solution are entirely my own work and not endorsed by Aerospike or Confluent. The code is PoC quality only and it is not production strength, and is available to anyone under the MIT License.

Near real-time Campaign Reporting Part 1 - Event Collection

Peter Milne — Mon, 13 Jan 2020 11:44:10 +0000

This is the first in a series of articles describing a simplified example of near real-time Ad Campaign reporting on a fixed set of campaign dimensions usually displayed for analysis in a user interface. The solution presented in this series relies on Kafka, Aerospike’s edge-to-core data pipeline technology, and Apollo GraphQL

Part 1: real-time capture of Ad events via Aerospike edge datastore and Kafka messaging.
Part 2: aggregation and reduction of Ad events via Aerospike Complex Data Type (CDT) operations into actionable Ad Campaign Key Performance Indicators (KPIs).
Part 3: describes how an Ad Campaign user interface displays those KPIs using GraphQL retrieve data stored in an Aerospike Cluster.

Data flow

This article, the code samples, and the example solution are entirely my own work and not endorsed by Aerospike. The code is available to anyone under the MIT License.

The use case — Part 1

A simplified use case for Part 1 here consists of capturing Ad events from Publishers, Advertisers and Vendors in an Aerospike edge datastore and publishing via Kafka

Impression sequence

Companion code

The companion code is in GitHub. The complete solution is in the master branch. The code for this article is in the ‘part-1’ branch.

Javascript and Node.js is used in each service although the same solution is possible in any language

The solution consists of:

A Publisher Simulator — Node.js
An Event Collector service — Node.js
Aerospike configurations enabling Kafka
Docker compose yml
Docker containers for:
- Aerospike Enterprise Edition (https://dockr.ly/2KZ6EUH)
- Zookeeper (https://dockr.ly/2KZgaaw)
- Kafka (https://dockr.ly/2L4NwVA)
- Kafka CLI (https://dockr.ly/2KXYEn4)

Docker and Docker Compose simplify the setup to allow you to focus on the Aerospike specific code and configuration.

What you need for the setup

Aerospike Enterprise Edition
An Aerospike user name and password
An Aerospike Feature Key File from Aerospike Support containing features:
- sdb-change-notification
- asdb-strong-consistency
- mesg-kafka-connector
Docker and Docker Compose

Setup steps

Step 1. Clone the GitHub repository with one of the following options

To get the whole repository use:

$ git clone (https://github.com/helipilot50/real-time-reporting-aerospike-kafka
$ cd real-time-reporting-aerospike-kafka
$ git checkout part-1

To get part-1 only use:

$ git clone --single-branch --branch part-1 https://github.com/helipilot50/real-time-reporting-aerospike-kafka

Step 2. Edit the file docker-compose.yml and add you Aerospike customer user name and password

Step 3. Copy your Feature Key File to project etc directory for the Edge and Core datastore; and to the edge exporter:

$ cp ~/features.conf <project-root>/aerospike/edge/etc/aerospike/features.conf
$ cp ~/features.conf <project-root>/aerospike/core/etc/aerospike/features.conf
$ cp ~/features.conf <project-root>/edge-exporter/features.conf

Step 4. Then run

$ docker-compose up

Once up and running, after the services have stabilized, you will see the output in the console similar to this:

Sample console output

How do the components interact?

Component Interaction

Docker Compose orchestrates the creation of several services in separate containers:

Zookeeper zookeeper — a single instance of Zookeeper to maintain the naming and configuration for Kafka (and other things)

Kafka kafka — a single node Kafka cluster

Kafka CLI kafkacli — Kafka CLI is used to view the messages in the Kafka topic

Aerospike Edge edge-aerospikedb — a single node Aerospike Enterprise cluster in Availability Mode. Availability is important to ensure every event is captured in real-time. The Dockerfile and other files for the Aerospike Enterprise server is located in the edge-aerospike directory. This container also mounts four volumes at runtime.

Edge container volume mounts

Aerospike Core core-aerospikedb — a single node Aerospike Enterprise cluster in Strong Consistency mode. Consistency is necessary as Campaign KPIs are used for payment a.k.a they represent money. The Dockerfile and other files for the Aerospike Enterprise server is located in the core-aerospike directory. This container also mounts three volumes at runtime.

Core container volume mounts

Edge Exporter edge-exporter — A single instance of the Aerospike Kafka Connector service. The edge-exporter directory contains the Dockerfile and feature.conf that define the connector service. This container also mounts two volumes at runtime

Kafka Connector volume mounts

Data Initializer data-initializer — A node.js program to initialize Campaign data and Tags in the core data store.

It runs briefly at the start of the docker-compose sequence, checks for data, writes sample data if none exist, and then terminates.

Publisher Simulator publisher-simulator — A Node.js program that produces random events to simulate the activity of publishers, advertisers and vendors. Each event is decorated with simulated Geo and User-agent data.

These events are generated with a completely contrived ratio and serve as an example only. In the real world, most events are impressions, with one click for every thousand impressions and one conversion in fifty clicks.

The publisher-simulator directory contains the simple node source, package.json and the Dockerfile.

Event Collector event-collector - A node.js REST API service implemented using Express. This service is a simple API that receives an Event from a Tag via a POST request and stores the event data in the edge datastore. The event-collector directory contains the node source, package.json and the Dockerfile.

Both the Event Collector and the Publisher Simulator use the Aerospike Node.js client. On the first build, both containers download and compile the supporting C library. The Dockerfile for both containers uses a multi-stage build that minimises the number of times the C library is compiled.

How is the solution deployed?

Each container is deployed using docker-compose on your local machine.

Deployment

How does the solution work?

Connecting to Aerospike

The code to connect to an Aerospike Cluster is similar in each component. You provide one or more address and ports in an array to the Aerospike.connect() method.

console.log('Attempting to connect to Aerospike cluster', asHost, asPort);

Aerospike.connect({
  hosts: [
    { addr: asHost, port: asPort }
  ],
  policies: {
    read: new Aerospike.ReadPolicy({
      totalTimeout: 100
    }),
    write: new Aerospike.WritePolicy({
      totalTimeout: 100
    }),
  },
  log: {
    level: Aerospike.log.INFO
  }
}).then(client => {
  asClient = client;
  console.log('Connected to aerospike', asHost, asPort);
}).catch(error => {
  console.error('Cannot connect to aerospike', error);
  throw error;
});

Connecting to Aerospike

The Aerospike client iterates through the array of IP addresses and ports until it successfully connects to a node. It then discovers all nodes in the cluster.

Data initializer

The data initializer creates 100 Campaigns with 1000 Tags per Campaign in the Core datastore. This forms the basic data for the entire simulation.

Data model

Campaigns are created with an Aerospike simpleput operation with Bins representing the indexable fields of a Campaign and a Complex Data Type (CDT) is initialized to form an elementary data cube for KPIs.

let campaignId = uuidv4();
// write campaign
let campaignKey = new Aerospike.Key(config.namespace, config.campaignSet, campaignId);
let bins = {};
bins[config.campaignIdBin] = campaignId;
bins[config.statsBin] = {
  clicks: 0,
  impressions: 0,
  visits: 0,
  conversions: 0,
};
bins[config.campaignNameBin] = `Acme campaign ${i}`;
await asClient.put(campaignKey, bins);
console.log('created campaign', bins[config.campaignNameBin]);

Campaign data

For each Campaign, one thousand Tags are created. A Tag is actually JavaScript tag added to a web page to reference the Creative and provide Ad events. Tags have unique Ids and are related to a Campaign via an Execution Plan. In this example, a Tag record is created with a reference to the Campaign.

tag = uuidv4();
// write tag-campaign mapping to aerospike
let tagKey = new Aerospike.Key(config.namespace, config.tagSet, tag);
let tagBins = {};
tagBins[config.tagIdBin] = tag;
tagBins[config.campaignIdBin] = campaignId;
await asClient.put(tagKey, tagBins);

Tag to Campaign mapping

For the simulator to send events with valid Tags, Aerospike is used as an associative array allowing an index lookup to retrieve a Tag Id, with a simple numeric index referencing the Tag Id.

let tagListKey = new Aerospike.Key(config.namespace, config.tagSet, tagIndex);
tagBins = {};
tagBins[config.tagIdBin] = tag;
await asClient.put(tagListKey, tagBins);

Indexing Tags with a numeric key

Publisher Simulator

The Publisher Simulator emits an Ad event on a defined interval to simulate the action of people interacting with an Ad. The interval between events deliberately large so we can see the whole sequence.

Every interval, the simulator:

Creates a random event
Reads a random Tag from the Campaign data stored in Aerospike
Simulates the Publisher/Advertiser/Vendor id
Sends the event to the event collector

const intervalFunc = async () => {
  try {

    let event = randomEvent();

    let index = Math.floor(Math.random() * 100000);
    let tagKey = new Aerospike.Key(config.namespace, config.tagSet, index);
    let record = await asClient.get(tagKey);
    let tag = record.bins[config.tagIdBin];

    let sourceId = uuidv4();
    let userAgent = randomUserAgent();
    let options = {
      uri: `http://event-collector:${PORT}/event/${event}`,
      headers: {
        'user-agent': userAgent,
      },
      json: {
        event: event,
        tag: tag,
        geo: randomGeo(),
      }
    };

    switch (event) {
      case 'click':
        options.json['publisher'] = sourceId;
        break;
      case 'impression':
        options.json['publisher'] = sourceId;
        break;
      case 'visit':
        options.json['advertiser'] = sourceId;
        break;
      case 'conversion':
        options.json['vendor'] = sourceId;
        break;
    }
    console.log(`Event:`, options)
    request.post(options);
  } catch (error) {
    console.error('send event error', error);
  }
}

setInterval(intervalFunc, INTERVAL);

Creating random event types with a known set of Tags

This simulator can be scaled up by changing the interval and running multiple containers.

Even collector

The Event Collector is a Web API implemented with Express.

Each event type has a specific route where the user agent is added to the body of the message and the writeEvent() method is called passing the body of the message.

const applyUserAgent = (req) => {
  req.body.userAgent = req.headers['user-agent'];
  console.log('event:', req.body);
};

EventRouter.route('/impression').post(function (req, res) {
  applyUserAgent(req);
  writeEvent('impression', req.body);
});

EventRouter.route('/visit').post(function (req, res) {
  applyUserAgent(req);
  writeEvent('visit', req.body);
});

EventRouter.route('/conversion').post(function (req, res) {
  applyUserAgent(req);
  writeEvent('conversion', req.body);
});

EventRouter.route('/click').post(function (req, res) {
  applyUserAgent(req);
  writeEvent('click', req.body);
});

app.use(bodyParser.json());
app.use(bodyParser.urlencoded({ extended: true }));
app.use('/event', EventRouter);

app.listen(PORT, () => {
  console.log(`Event Collector running on port ${PORT}`);
});

Web API using Express

Events are received from the Publisher/Vendor/Advertiser and stored in an Aerospike edge data store. This store acts as high availability and low latency buffer between the event collector and Kafka.

The raw event is classified by assigning some elements of the event to Bins and then stored using a put operation.

const writeEvent = async (type, body) => {
  const eventId = uuidv4();
  try {

    ... connecto to Aerospike here exactly once ...

    let clickKey = new Aerospike.Key(config.namespace, config.eventsSet, eventId);
    let bins = {};
    bins[config.eventIdBin] = eventId;
    bins[config.eventBin] = body;
    bins[config.tagBin] = body.tag;
    switch (type) {
      case 'click':
        bins[config.sourceBin] = body.publisher;
        break;
      case 'impression':
        bins[config.sourceBin] = body.publisher;
        break;
      case 'visit':
        bins[config.sourceBin] = body.advertiser;
        break;
      case 'conversion':
        bins[config.sourceBin] = body.vendor;
        break;
    }
    bins[config.typeBin] = type;
    await asClient.put(clickKey, bins);
    console.log(`Processed ${type} event`, eventId);
  } catch (error) {
    console.error(`${type} event processing error`, eventId);
    throw error;
  }
}

Storing events in Aerospike

Edge Exporter

The Edge Exporter is the Aerospike outbound Kafka connector running in a container. Each time a record is written to Aerospike it is exported to Kafka based on setting in the aerospike.conf file on each node in the Aerospike cluster.

xdr {
    enable-xdr true
    enable-change-notification true
    xdr-digestlog-path /etc/aerospike/digestlog.log1 100G
    xdr-compression-threshold 1000
    xdr-info-port 3004
    datacenter outbound-kafka {
            dc-type http
            http-version v2
            http-url http://edge-exporter:8080/aerospike/kafka/publish
    }
}

aerospike.conf

To enable Aerospike to export to Kafka, a features.conf file is required to be available to both the Aerospike Kafka Connector and the Aerospike Cluster.

# generated 2019-07-19 17:15:53

feature-key-version              1
serial-number                    291903886

valid-until-date                 2019-10-18

asdb-change-notification         true
mesg-kafka-connector             true
raf-realtime-analysis-framework  true

----- SIGNATURE ------------------------------------------------
<signature here>
----- END OF SIGNATURE -----------------------------------------

features.conf

Scaling the solution

How fast will it be? -This depends on the technology and hardware used.

Ad Event data is captured in real-time in the edge Aerospike cluster. Aerospike scales horizontally by adding nodes to the cluster. In fact, Aerospike’s ability to scale is one of its most powerful features. Likewise, Kafka is also designed to scale easily. Both technologies have extensive documentation and guides on scaling for throughput, latency, availability and capacity.

Aerospike and Kafka go hand-in-glove. Some of the benefits of Aerospike as a front-end to Kafka are as follows:

Aerospike gives you a high-throughput, low-latency, datastore configurable to prioritise high-availability or strong consistency at a lower TOC with Flash memorySSDs.
Ad Event data is captured in real-time, so Kafka service needs less capacity than otherwise.
Ad Events are log-level data stored in Aerospike. These data can thus be used for even more analytics with the Aerospike Spark Connector or other analytics tools.

The Aerospike Kafka Connector runs in the Jetty web server, and can be “Dockerized” and scaled like any other microservice.

For microservices in Docker containers, Kubernetes is my favourite way to orchestrate for production with excellent autoscaling and high availability features and several CI/CD tools integrate directly with it.

Orchestration with Kubernetes and tools like Drone and Helm Charts, enables your Developers to focus on development, and your DevOps not to hate the Developers.

Review

Part 1 (this article) of this series describes:

creating mock Campaign data
a publisher simulator
an event receiver
an edge database
an edge exporter

Part 2 describes the aggregation and reduction of Ad events into Campaign KPIs using Kafka as the messaging system and Aerospike as the consistent data store.

Part 3 describes the Campaign service and Campaign UI for a user to view the Campaign KPIs in near real-time.

Forem: Peter Milne

Near Real-Time Campaign Reporting Part 3 - Campaign Service and Campaign UI

Summary of Part 1 and Part 2

The use case — Part 3

Companion code

What you need for the setup

Setup steps

How do the components interact?

How is the solution deployed?

How does the solution work?

Campaign service

index.js

Schema definition

Resolvers

CampaignDataSource.js

Fetching a single record by ID

Fetching multiple records an array of IDs

Fetching multiple records using a query

Transforming a record to a Campaign

KpiReceiver

Campaign UI

index.js

CampaignList

CampaignRow

Kpi

The whole story

Event sequence

The complete component interaction

Review

Possible improvements to the entire application

Disclaimer

Near real-time Campaign Reporting Part 2 - Aggregation/Reduction

Summary of Part 1

The use case — Part 2

Companion code

What you need for the setup

Setup steps

How do the components interact?

How is the solution deployed?

How does the solution work?

Connecting to Kafka

Extract the event data

Lookup Campaign Id using Tag

Aggregating the Event

Publishing the new KPI

Review

Disclaimer

Near real-time Campaign Reporting Part 1 - Event Collection

The use case — Part 1

Companion code

What you need for the setup

Setup steps

How do the components interact?

How is the solution deployed?

How does the solution work?

Connecting to Aerospike

Data initializer

Publisher Simulator

Even collector

Edge Exporter

Scaling the solution

Review

Related Articles