alternative_sparkjob.py

logs = spark.read.json("in/logs/*.json.gz").createOrReplaceTempView('logs')
users = spark.read.json('in/user_hostname_mapping.json').createOrReplaceTempView('users')

billing_stats = spark.sql("""
    SELECT user,
           sum(size) as total_bandwidth
      FROM logs
INNER JOIN users ON logs.hostname = users.hostname
  GROUP BY user""")

billing_stats.coalesce(1).write.json('out/billing_stats')

in_logs_log-1.json

{"hostname": "kissa.com", "path": "/lol", "size": 12838}
{"hostname": "kissa.com", "path": "/lol", "size": 12838}
{"hostname": "kissa.com", "path": "/bal", "size": 12838}
{"hostname": "kissa.com", "path": "/bal", "size": 12838}
{"hostname": "karmes.com", "path": "/bal", "size": 4222}
{"hostname": "karmes.com", "path": "/bal", "size": 4222}
{"hostname": "karmes.com", "path": "/bal", "size": 4222}
{"hostname": "karmes.com", "path": "/bal", "size": 4222}
{"hostname": "karmes.com", "path": "/lol", "size": 4222}
{"hostname": "karmes.com", "path": "/lol", "size": 4222}
{"hostname": "karmes.com", "path": "/lol", "size": 4222}
{"hostname": "karmes.com", "path": "/lol", "size": 12838}
{"hostname": "karmes.com", "path": "/lol", "size": 12838}
{"hostname": "kissa.com", "path": "/lol", "size": 12838}
{"hostname": "kissa.com", "path": "/lol", "size": 12838}
{"hostname": "kissa.com", "path": "/lol", "size": 12838}
{"hostname": "kissa.com", "path": "/lol", "size": 12838}
{"hostname": "katti.com", "path": "/hei", "size": 8732}
{"hostname": "katti.com", "path": "/hei", "size": 8732}

in_user_hostname_mapping.json

{"user": "jamo", "hostname": "katti.com"}
{"user": "jamo", "hostname": "kissa.com"}
{"user": "karmemies", "hostname": "karmes.com"}
{"user": "joni", "hostname": "vuohi.com"}

sparkjob.py

# Read compressed logs
logs = spark.read.json("in/logs/*.json.gz")
# Read user-hostname mapping
users = spark.read.json('in/user_hostname_mapping.json')
# 1. Join user-mappings
# 2. Group by 'user'
# 3. Aggregate bandwidth_total from 'size' of each user requests
billing_stats = logs.join(users, 'hostname').groupBy('user').agg(F.sum('size').alias('total_bandwidth'))
# Write out in single JSON-file
billing_stats.coalesce(1).write.json('out/billing_stats')