import sys

from pyspark.sql import SparkSession
from pyspark.sql import functions as F


if len(sys.argv) != 3:
    raise SystemExit("Usage: sg_spark_submit.py <input-path> <output-path>")

input_path = sys.argv[1]
output_path = sys.argv[2]

spark = SparkSession.builder.appName("sg-spark-submit").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

events = spark.read.csv(
    input_path,
    schema="event STRING, count INT",
)

summary = events.groupBy("event").agg(F.sum("count").alias("total")).orderBy("event")
summary.coalesce(1).write.mode("overwrite").json(output_path)

for row in summary.collect():
    print(f"{row['event']}={row['total']}")

print(f"application_id={spark.sparkContext.applicationId}")
print(f"output_path={output_path}")

spark.stop()