App.elm

import Stream exposing (Stream, Table)
import Json.Decode

type alias Username = String
type alias Clicks = Integer
type alias Region = String

userClicks : Stream Username Clicks
userClicks = Stream.stream "user-clicks"
  |> Stream.mapValue String.toInt
  |> Stream.dropErrors

userRegions : Table Username Region
userRegions = Stream.table "user-regions"

clicksPerRegion : Table Region Clicks
clicksPerRegion =
  userClicks
  |> Stream.leftJoin (always "n/a") userRegions
  |> Stream.map \username (region, clicks) -> (region, clicks)
  |> Stream.reduce (+)

main : Task Stream.Error ()
main =
  Stream.write "clicks-per-region" clicksPerRegion

Stream.elm

module Stream exposing (..)

type Stream k v = Stream
type Table k v = Table

stream : String -> Stream String String

table : String -> Table String String

decodeKey : (String -> Result String k) -> Stream String v -> Stream k v

decodeValue : (String -> Result String v) -> Stream k String -> Stream k v
decodeValue f stream =
  Stream.map f stream
  |> Stream.dropErrors

-- Drops errors from a stream.
dropErrors : Stream k (Result err v) -> Stream k v

leftJoin : (k -> v1) -> Table k v1 -> Stream k v2 -> Stream k (v1, v2)

map : ((k, v) -> (k', v')) -> Stream k v -> Stream k' v'

reduce : ((v, v) -> v) -> Stream k v -> Table k v