Build Your First App Pipeline

Use these steps to load the App Flow Data Feed and the App Event Data Feed, compute flow and event metrics, join the two feeds, and check the results against Conviva Pulse. Read App Data Feeds: Start Here before you begin.

Both feeds deliver their metrics and metadata inside JSON columns. Step 4 shows how to read them, and every later step depends on it.

Step 1: Set up delivery

Pick a destination and a format for each feed you subscribe to. Conviva Connect writes the files to your storage:

• Amazon S3, including the AWS role for KMS-encrypted uploads

• Google Cloud Storage

• SFTP

Choose Parquet or Avro where you can. Both preserve MetricsInt, MetricsFloat, Tags and EventDetails as native map types. CSV serializes each of them into a single JSON string, which you parse on read as shown in step 4.

On the Flow Feed you also choose which flows you receive. You can take every flow configured for your account, or name a subset by flow ID and flow name. Confirm your configured flow list and your column list with your Conviva Representative before you write the table definition, and ask to be notified when either changes.

Step 2: What each delivery window contains

The two feeds are delivered on different schedules.

The Flow Feed is delivered every minute. Each row is one instance of one configured flow, identified by FlowId. Conviva writes the row once, in the file for the minute the flow closed, either because it completed or because a session or flow timeout ended it. A flow can stay open much longer than one minute: in the delivered sample, four flows started 34 minutes before the file that carries them.

The Event Feed is delivered hourly. Each event appears in exactly one hourly file. Each row is one event, stamped with EventTimeMs and tied to an app session through ECOSessionId. An app session closes after 90 seconds with no user interaction, so a single device produces many sessions across an hour.

For each window, Conviva writes the data into a dated folder containing multiple part files and a .manifest file that lists them. Read the manifest first and load only the files it lists, so you never pick up a partially written folder. Load every part file in a folder as one logical dataset for that window.

Neither App feed carries a dt partition column. The window is carried by the folder, so record it yourself as you load, and partition your target table on the value you derive.

Step 3: Load the raw files

Land both feeds unchanged in raw tables first and transform downstream. With the raw rows kept, you can rebuild the downstream tables after you correct a parsing error.

CREATE TABLE app_flow_raw (
  FlowId            STRING,
  Flow              STRING,
  UserId            STRING,
  ClientId          STRING,
  SensorVersion     STRING,
  Country           STRING,
  Platform          STRING,
  DeviceCategory    STRING,
  AppName           STRING,
  AppVersion        STRING,
  StartTimeStampMs  BIGINT,
  EndTimeStampMs    BIGINT,
  MetricsInt        MAP<STRING, BIGINT>,
  MetricsFloat      MAP<STRING, DOUBLE>,
  Tags              MAP<STRING, STRING>
  -- plus the rest of your configured columns
);

CREATE TABLE app_event_raw (
  ECOSessionId   STRING,
  EventTimeMs    BIGINT,
  ClientId       STRING,
  UserId         STRING,
  Country        STRING,
  IpV4           STRING,
  IpV6           STRING,
  ConnectionType STRING,
  EventCategory  STRING,
  EventName      STRING,
  EventDetails   MAP<STRING, STRING>
);

In Parquet and Avro the four JSON columns arrive as maps and the MAP types above are literal. In CSV they arrive as JSON strings, so declare them STRING and parse them in step 4. See the schema and column dictionary for the type and mode of every field.

Step 4: Read the JSON columns

Four columns hold JSON objects: MetricsInt and MetricsFloat hold the flow's measures, Tags holds your own flow metadata, and EventDetails holds your own event metadata. Their keys vary from row to row.

A key that does not apply to a row is missing from the object. Extracting it returns null, so wrap it in COALESCE whenever a missing key should count as zero.

Every query from here on reads the CSV form, where the column is a JSON string:

SELECT
  FlowId,
  Flow,
  CAST(JSON_EXTRACT_SCALAR(MetricsInt, '$.success')           AS BIGINT) AS success,
  CAST(JSON_EXTRACT_SCALAR(MetricsInt, '$.failure')           AS BIGINT) AS failure,
  CAST(JSON_EXTRACT_SCALAR(MetricsInt, '$.error')             AS BIGINT) AS error_count,
  CAST(JSON_EXTRACT_SCALAR(MetricsInt, '$.total_duration_ms') AS BIGINT) AS total_duration_ms
FROM app_flow_raw;

In Parquet and Avro the same column is a map, so you read it with a key lookup, MetricsInt['success'], and the CAST goes away. Everything the rest of this page says about missing keys and sentinels applies to both forms.

MetricsInt holds a different set of keys depending on how the row is configured in Conviva. The delivered sample carries three kinds of row across its 14 rows.

Standard flows carry success and failure as a pair of 0 or 1 flags, exactly one of which is set, plus total_duration_ms. They can also carry error and the network request keys. Seven of the sample rows are standard flows.

Event aggregation metrics carry keys named after the metric. A Count writes one integer key. The sample carries five Counts, each with a value of 1: Video Attempt twice, plus Tile Clicks, Playback Initiated and Video Play. Sum and Average results go to MetricsFloat.

Duration metrics carry total_duration_ms and nothing else, and Conviva emits them only for positive durations. The sample carries two, Video Attempt to Stream Auth Request at 130 ms and Video Attempt to Manifest Request at 648 ms.

Two more kinds can appear in your feed. A conversion metric carries success and failure with no duration. A row whose Flow value is session is an app session and carries the session metric keys. Metric calculations lists them.

Your flow configuration decides which kind each row is, so filter on Flow names where you know them. Otherwise use the keys: standard flows are the only rows that carry both success and total_duration_ms. Every flow metric on this page uses that filter. Without it, the denominator includes rows that have no outcome, and the rates come out wrong.

MetricsFloat uses the same encoding for floating-point measures, including event aggregation Sum and Average results. An Average writes <name>_count to MetricsInt and <name>_value, already averaged for that row, to MetricsFloat. The delivered sample has no Sum or Average metrics, so MetricsFloat is empty on every row.

Tags and EventDetails carry the metadata your own app reports. Pivot the keys you report on into real columns once, at load time, rather than extracting them on every query. Prefer a dedicated column over a tag when both carry the same thing, because the dedicated column is typed and the tag is always a string.

Step 5: Assign flows to a reporting period

Conviva writes each flow instance once, when it closes, so the Flow Feed needs no deduplication. FlowId is generated from the ClientId, the flow name and the start time, so two runs of the same flow get different values.

The file a row arrives in tells you when the flow closed. To count the flows that started in a period, group on StartTimeStampMs:

SELECT
  DATE_TRUNC('hour', FROM_UNIXTIME(StartTimeStampMs / 1000)) AS start_hour,
  Flow,
  COUNT(*) AS flow_init
FROM app_flow_raw
WHERE JSON_EXTRACT_SCALAR(MetricsInt, '$.success') IS NOT NULL
  AND JSON_EXTRACT_SCALAR(MetricsInt, '$.total_duration_ms') IS NOT NULL
GROUP BY 1, 2;

A flow that starts at 10:59 and completes at 11:01 arrives in the 11:01 file. So recompute each period as later files arrive, because flows that started in it keep arriving until they close.

Step 6: Roll up flow outcomes

This roll-up produces the flow metrics you see in Pulse: Flow Init, Flow Complete, Flow Success, Flow Error and Flow Avg Duration. It groups by flow only; add a period from step 5 to report over time.

WITH m AS (
  SELECT
    Flow,
    CAST(JSON_EXTRACT_SCALAR(MetricsInt, '$.success')           AS BIGINT) AS success,
    CAST(JSON_EXTRACT_SCALAR(MetricsInt, '$.error')             AS BIGINT) AS error_count,
    CAST(JSON_EXTRACT_SCALAR(MetricsInt, '$.total_duration_ms') AS BIGINT) AS duration_ms
  FROM app_flow_raw
  WHERE JSON_EXTRACT_SCALAR(MetricsInt, '$.success') IS NOT NULL
    AND JSON_EXTRACT_SCALAR(MetricsInt, '$.total_duration_ms') IS NOT NULL
)
SELECT
  Flow,
  COUNT(*)                                             AS flow_init,
  SUM(CASE WHEN duration_ms >= 0 THEN 1 ELSE 0 END)    AS flow_complete,
  SUM(success)                                         AS flow_success,
  SUM(COALESCE(error_count, 0))                        AS flow_error,
  100.0 * SUM(success) / COUNT(*)                      AS flow_success_rate_pct,
  AVG(CASE WHEN duration_ms >= 0 THEN duration_ms END) AS flow_avg_duration_ms
FROM m
GROUP BY Flow;

The query depends on three details. Removing any of them produces wrong results.

The WHERE clause keeps standard flows only. Every other kind of row is missing success, total_duration_ms or both, and including them would put rows with no outcome into the denominators.

Flow Complete and Flow Success are counted differently. success is 1 only when the flow completed within its configured success threshold. A flow that completes after the threshold still has a duration of 0 or more, but carries success of 0 and failure of 1. So the query counts Flow Complete from the duration and Flow Success from success.

100.0 forces the division to floating point. Trino, Presto, Hive and Spark all truncate integer division, so SUM(success) / COUNT(*) returns 0 for any flow that doesn't succeed every single time. Keep a decimal literal in the numerator of every ratio you write.

The CASE expressions exclude the duration sentinels. A negative total_duration_ms means a timeout closed the flow before it completed: -1 when the app session timed out, -2 when the flow's completion timeout was reached. Treat both as status codes. Across the seven standard flows in the delivered sample, an unguarded AVG(total_duration_ms) returns 848 ms against a true average of 1,980 ms, because four of the seven carry -1. Exclude sentinels from ratios as well as from sums and averages. A flow whose every instance timed out returns a null average, because no instance completed.

Step 7: Roll up events

The Event Feed has no metric columns. Every measure you want is a count, a rate, or a time span you compute yourself over EventCategory, EventName and the keys inside EventDetails.

SELECT
  ECOSessionId,
  ClientId,
  UserId,
  COUNT(*)                                                             AS events,
  SUM(CASE WHEN EventCategory = 'Frustration Signal' THEN 1 ELSE 0 END) AS frustration_events,
  MIN(EventTimeMs)                                                     AS first_event_ms,
  MAX(EventTimeMs)                                                     AS last_event_ms
FROM app_event_raw
GROUP BY ECOSessionId, ClientId, UserId;

EventCategory groups events into a small set that Conviva assigns. The delivered sample carries Custom, Media, Page, Search, Payment and Frustration Signal. Treat it as a growing set rather than a fixed list, and write your roll-ups so a new category appears in the output instead of being silently dropped.

EventName mixes two populations. Names prefixed with conviva_, such as conviva_page_view and conviva_video_error, are generated by the Conviva Sensor. Everything else is a name your own app reports. Report on the two groups separately.

Step 8: Join the Flow Feed to the Event Feed

The two feeds share no session key. The Flow Feed has no ECOSessionId, and the Event Feed has no FlowId. Join them on identity instead: ClientId identifies the device and UserId identifies the user, and the pair ties a flow to the events that happened around it.

SELECT
  f.FlowId,
  f.Flow,
  COUNT(e.EventName)              AS events_in_window,
  COUNT(DISTINCT e.EventCategory) AS event_categories
FROM app_flow_raw f
LEFT JOIN app_event_raw e
  ON  e.ClientId = f.ClientId
  AND e.UserId   = f.UserId
  AND e.EventTimeMs BETWEEN f.StartTimeStampMs AND f.EndTimeStampMs
GROUP BY f.FlowId, f.Flow;

Use a left join. An inner join drops flows that have no matching events, and a left join keeps them with a count of zero.

Size the time window to your flows. The predicate above bounds events to the flow's own start and end, which is right for a short flow and too narrow for a long one. Widen it when you want the lead-up to a flow, and remember that a flow closed by a timeout reports no duration, so its start and end timestamps are the only window you have.

Align the windows before you join. The Flow Feed arrives every minute and the Event Feed arrives hourly, so a flow is usually loadable well before the hour of events around it has been delivered. Join only on hours for which both feeds have been delivered.

Step 9: Validate against Pulse

Reconcile your output against Conviva Pulse before you publish it. Pick a day, a flow, and a metric (for example, Flow Complete for one configured flow), compute it from your loaded data, and compare it to the same view in Pulse.

If a metric differs significantly from Pulse, check the following:

• Did a missing key get read as zero instead of null, or the reverse? A success rate that is too low often points to this.

• Did an integer division truncate a ratio to 0 or 1?

• Did a -1 or -2 duration sentinel leak into an average?

• Did you count Flow Complete with SUM(success)? That measures Flow Success.

• Did event aggregation, duration, conversion or session rows land in a denominator that should hold standard flows only?

• Did you load every window, including the files that arrived after your period closed? The Flow Feed delivers 1,440 folders a day, and each missing folder removes the flows it carries.

Next steps

• Schema and column dictionary for every field in both feeds.

• Metric calculations for the flow and session metric definitions.

• Choose your feed if you are still deciding which feed answers your question.