Parses kafka group assignments and outputs in more readable form

parse_partition_assignments.py

#!/usr/bin/env python
import sys

def parse_group_assignments():
	data = sys.stdin.read()
	lines = data.splitlines()[2:]
	structured = []
	for line in lines:
		cols = [chunk for chunk in line.split(" ") if len(chunk) > 0]
		structured.append({"group_id": cols[0],
			"topic": cols[1],
			"partition": cols[2],
			"offset": cols[3],
			"messages": cols[4],
			"lag": cols[5],
			"consumer_id": cols[6],
			"host": cols[7],
			"client_id": cols[8]})
	return structured

def details_for_host(assignments, host):
	host_assignments = [a for a in assignments if a["host"] == host]
	# extract thread names from assingments to order by
	client_ids = set([a['client_id'] for a in host_assignments])
	threads = {client_id: {"thread_id": client_id.split("-StreamThread-")[1][:1], "assignments": []}  for client_id in client_ids}
	for a in host_assignments:
		threads[a["client_id"]]["assignments"].append(a)

	return {"host": host, 
		"threads": threads}

assignments = parse_group_assignments()
# print assignments
hosts = set([c["host"] for c in assignments])
host_details = [details_for_host(assignments, h) for h in hosts]
#print host_details

for host in host_details:
	thread_assignments = host["threads"].values()
	client_id = thread_assignments[0]["assignments"][0]["client_id"].split("-StreamThread")[0]
	print "Host: %s, %s" % (host["host"], client_id)
	print "\tThreads:"
	for t in sorted(thread_assignments):
		topics = [a["topic"] + ", partition: " + a["partition"] + ", lag: " + a["lag"] for a in t["assignments"]]
		print "\t\t%s: %s" % (t["thread_id"], topics)
	print ""