TEZ-4666: Migrate tez-tools python scripts to python3

Author: Aggarwal-Raghav

tez-tools/counter-diff/README.md

@@ -26,11 +26,11 @@ Before using the tool, make sure to install texttable using `pip install texttab
 
 To use the tool, run
 
-`python counter-diff.py dag_1.zip dag_2.zip [--detail]`
+`python3 counter-diff.py dag_1.zip dag_2.zip [--detail]`
 
 This will print counter output difference between the specified DAGs like this
 
-Example: `python counter-diff.py dag_1499978510619_0002_143.zip dag_1499978510619_0002_144.zip`
+Example: `python3 counter-diff.py dag_1499978510619_0002_143.zip dag_1499978510619_0002_144.zip`
 
 ```
 +-------------------+-------------------------------------+----------------------------+----------------------------+-------------+
@@ -81,4 +81,4 @@ Example: `python counter-diff.py dag_1499978510619_0002_143.zip dag_149997851061
 |                   | KILLED_TASKS                        | 0                          | 0                          | 0           |
 |                   | TIME_TAKEN                          | 250198                     | 68981                      | -181217     |
 +-------------------+-------------------------------------+----------------------------+----------------------------+-------------+
-```
+```

tez-tools/counter-diff/counter-diff.py

@@ -17,13 +17,18 @@
 # under the License.
 #
 
-import imp, json, os, shutil, sys, tempfile, zipfile
+import json
+import os
+import shutil
+import sys
+import tempfile
+import zipfile
+
 try:
-    imp.find_module('texttable')
     from texttable import Texttable
 except ImportError:
-	sys.stderr.write("Could not import Texttable\nRetry after 'pip install texttable'\n")
-	exit()
+    print("Could not import Texttable. Retry after 'pip install texttable'", file=sys.stderr)
+    sys.exit(1)
 
 tmpdir = tempfile.mkdtemp()
 
@@ -47,28 +52,33 @@ def diff(file1, file2):
 	# also in dag.json data is inside "dag" root node
 	file1_using_dag_json = True
 	dag_json_file1 = os.path.join(file1_dir, "dag.json")
-	if os.path.isfile(dag_json_file1) == False:
+	if not os.path.isfile(dag_json_file1):
 		file1_using_dag_json = False
 		dag_json_file1 = os.path.join(file1_dir, "TEZ_DAG")
-		if os.path.isfile(dag_json_file1) == False:
-			print "Unable to find dag.json/TEZ_DAG file inside the archive " + file1
-			exit()
+		if not os.path.isfile(dag_json_file1):
+			print("Unable to find dag.json/TEZ_DAG file inside the archive " + file1)
+			sys.exit()
 
 	file2_using_dag_json = True
 	dag_json_file2 = os.path.join(file2_dir, "dag.json")
-	if os.path.isfile(dag_json_file2) == False:
+	if not os.path.isfile(dag_json_file2):
 		file2_using_dag_json = False
 		dag_json_file2 = os.path.join(file2_dir, "TEZ_DAG")
-		if os.path.isfile(dag_json_file2) == False:
-			print "Unable to find dag.json/TEZ_DAG file inside the archive " + file1
-			exit()
+		if not os.path.isfile(dag_json_file2):
+			print("Unable to find dag.json/TEZ_DAG file inside the archive " + file1)
+			sys.exit()
 
 	# populate diff table
 	difftable = {}
 	with open(dag_json_file1) as data_file:
 		file1_dag_json = json.load(data_file)["dag"] if file1_using_dag_json else json.load(data_file)
-		counters = file1_dag_json['otherinfo']['counters']
-		for group in counters['counterGroups']:
+		
+		# Safe access to otherinfo and counters
+		otherinfo = file1_dag_json.get('otherinfo', {})
+		counters = otherinfo.get('counters', {})
+		
+		# Iterate only if counterGroups exists
+		for group in counters.get('counterGroups', []):
 			countertable = {}
 			for counter in group['counters']:
 				counterName = counter['counterName']
@@ -78,22 +88,24 @@ def diff(file1, file2):
 			groupName = group['counterGroupName']
 			difftable[groupName] = countertable
 
-		# add other info
-		otherinfo = file1_dag_json['otherinfo']
+		# add other info safely
 		countertable = {}
-		countertable["TIME_TAKEN"] = [otherinfo['timeTaken']]
-		countertable["COMPLETED_TASKS"] = [otherinfo['numCompletedTasks']]
-		countertable["SUCCEEDED_TASKS"] = [otherinfo['numSucceededTasks']]
-		countertable["FAILED_TASKS"] = [otherinfo['numFailedTasks']]
-		countertable["KILLED_TASKS"] = [otherinfo['numKilledTasks']]
-		countertable["FAILED_TASK_ATTEMPTS"] = [otherinfo['numFailedTaskAttempts']]
-		countertable["KILLED_TASK_ATTEMPTS"] = [otherinfo['numKilledTaskAttempts']]
+		countertable["TIME_TAKEN"] = [otherinfo.get('timeTaken', 0)]
+		countertable["COMPLETED_TASKS"] = [otherinfo.get('numCompletedTasks', 0)]
+		countertable["SUCCEEDED_TASKS"] = [otherinfo.get('numSucceededTasks', 0)]
+		countertable["FAILED_TASKS"] = [otherinfo.get('numFailedTasks', 0)]
+		countertable["KILLED_TASKS"] = [otherinfo.get('numKilledTasks', 0)]
+		countertable["FAILED_TASK_ATTEMPTS"] = [otherinfo.get('numFailedTaskAttempts', 0)]
+		countertable["KILLED_TASK_ATTEMPTS"] = [otherinfo.get('numKilledTaskAttempts', 0)]
 		difftable['otherinfo'] = countertable
 
 	with open(dag_json_file2) as data_file:
 		file2_dag_json = json.load(data_file)["dag"] if file2_using_dag_json else json.load(data_file)
-		counters = file2_dag_json['otherinfo']['counters']
-		for group in counters['counterGroups']:
+		
+		otherinfo = file2_dag_json.get('otherinfo', {})
+		counters = otherinfo.get('counters', {})
+		
+		for group in counters.get('counterGroups', []):
 			groupName = group['counterGroupName']
 			if groupName not in difftable:
 				difftable[groupName] = {}
@@ -105,16 +117,15 @@ def diff(file1, file2):
 					countertable[counterName] = [0]
 				countertable[counterName].append(counter['counterValue'])
 
-		# append other info
-		otherinfo = file2_dag_json['otherinfo']
+		# append other info safely
 		countertable = difftable['otherinfo']
-		countertable["TIME_TAKEN"].append(otherinfo['timeTaken'])
-		countertable["COMPLETED_TASKS"].append(otherinfo['numCompletedTasks'])
-		countertable["SUCCEEDED_TASKS"].append(otherinfo['numSucceededTasks'])
-		countertable["FAILED_TASKS"].append(otherinfo['numFailedTasks'])
-		countertable["KILLED_TASKS"].append(otherinfo['numKilledTasks'])
-		countertable["FAILED_TASK_ATTEMPTS"].append(otherinfo['numFailedTaskAttempts'])
-		countertable["KILLED_TASK_ATTEMPTS"].append(otherinfo['numKilledTaskAttempts'])
+		countertable["TIME_TAKEN"].append(otherinfo.get('timeTaken', 0))
+		countertable["COMPLETED_TASKS"].append(otherinfo.get('numCompletedTasks', 0))
+		countertable["SUCCEEDED_TASKS"].append(otherinfo.get('numSucceededTasks', 0))
+		countertable["FAILED_TASKS"].append(otherinfo.get('numFailedTasks', 0))
+		countertable["KILLED_TASKS"].append(otherinfo.get('numKilledTasks', 0))
+		countertable["FAILED_TASK_ATTEMPTS"].append(otherinfo.get('numFailedTaskAttempts', 0))
+		countertable["KILLED_TASK_ATTEMPTS"].append(otherinfo.get('numKilledTaskAttempts', 0))
 		difftable['otherinfo'] = countertable
 
 	# if some counters are missing, consider it as 0 and compute delta difference
@@ -134,7 +145,7 @@ def print_table(difftable, name1, name2, detailed=False):
 	table = Texttable(max_width=0)
 	table.set_cols_align(["l", "l", "l", "l", "l"])
 	table.set_cols_valign(["m", "m", "m", "m", "m"])
-	table.add_row(["Counter Group", "Counter Name", name1, name2, "delta"]);
+	table.add_row(["Counter Group", "Counter Name", name1, name2, "delta"])
 	for k in sorted(difftable):
 		# ignore task specific counters in default output
 		if not detailed and ("_INPUT_" in k or "_OUTPUT_" in k):
@@ -177,13 +188,13 @@ def print_table(difftable, name1, name2, detailed=False):
 
 		table.add_row(row)
 
-	print table.draw() + "\n"
+	print(table.draw() + "\n")
 
 
 def main(argv):
 	sysargs = len(argv)
 	if sysargs < 2:
-		print "Usage: python counter-diff.py dag_file1.zip dag_file2.zip [--detail]"
+		print("Usage: python3 counter-diff.py dag_file1.zip dag_file2.zip [--detail]")
 		return -1
 
 	file1 = argv[0]
@@ -200,4 +211,4 @@ def main(argv):
 	try:
 		sys.exit(main(sys.argv[1:]))
 	finally:
-		shutil.rmtree(tmpdir)
+		shutil.rmtree(tmpdir)

tez-tools/requirements.txt

@@ -0,0 +1 @@
+texttable==1.7.0

tez-tools/swimlanes/amlogparser.py

@@ -17,14 +17,12 @@
 # under the License.
 #
 
-import sys,re
+import sys
+import re
 from itertools import groupby
-from bz2 import BZ2File
-from gzip import GzipFile as GZFile
-try:
-	from urllib.request import urlopen
-except:
-	from urllib2 import urlopen as urlopen
+import bz2
+import gzip
+from urllib.request import urlopen
 
 class AMRawEvent(object):
 	def __init__(self, ts, dag, event, args):
@@ -39,7 +37,7 @@ def first(l):
 	return (l[:1] or [None])[0]
 
 def kv_add(d, k, v):
-	if(d.has_key(k)):
+	if(k in d):
 		oldv = d[k]
 		if(type(oldv) is list):
 			oldv.append(v)
@@ -50,7 +48,7 @@ def kv_add(d, k, v):
 		d[k] = v
 
 def csv_kv(args):
-	kvs = {};
+	kvs = {}
 	pairs = [p.strip() for p in args.split(",")]
 	for kv in pairs:
 		if(kv.find("=") == -1):
@@ -160,10 +158,11 @@ def __repr__(self):
 
 class Attempt(object):
 	def __init__(self, pair):
-		start = first(filter(lambda a: a.event == "TASK_ATTEMPT_STARTED", pair))
-		finish = first(filter(lambda a: a.event == "TASK_ATTEMPT_FINISHED", pair))
+		# Consuming iterators immediately with list() for Py3 compatibility
+		start = first(list(filter(lambda a: a.event == "TASK_ATTEMPT_STARTED", pair)))
+		finish = first(list(filter(lambda a: a.event == "TASK_ATTEMPT_FINISHED", pair)))
 		if start is None or finish is None:
-			print [start, finish];
+			print([start, finish])
 		self.raw = finish
 		self.kvs = csv_kv(start.args)
 		if finish is not None:
@@ -186,26 +185,28 @@ def __repr__(self):
 
 def open_file(f):
 	if(f.endswith(".gz")):
-		return GZFile(f)
+		return gzip.open(f, "rt")
 	elif(f.endswith(".bz2")):
-		return BZ2File(f)
+		return bz2.open(f, "rt")
 	elif(f.startswith("http://")):
 		return urlopen(f)
-	return open(f)
+	return open(f, "r")
 
 class AMLog(object):	
 	def init(self):
 		ID=r'[^\]]*'
 		TS=r'[0-9:\-, ]*'
 		MAIN_RE=r'^(?P<ts>%(ts)s) [?INFO]? [(?P<thread>%(id)s)] \|?((HistoryEventHandler.criticalEvents)|((org.apache.tez.dag.)?history.HistoryEventHandler))\|?: [HISTORY][DAG:(?P<dag>%(id)s)][Event:(?P<event>%(id)s)]: (?P<args>.*)'
-		MAIN_RE = MAIN_RE.replace('[','\[').replace(']','\]')
+		# Fix for SyntaxWarning: using raw strings
+		MAIN_RE = MAIN_RE.replace('[', r'\[').replace(']', r'\]')
 		MAIN_RE = MAIN_RE % {'ts' : TS, 'id' : ID}
 		self.MAIN_RE = re.compile(MAIN_RE)
 
 	def __init__(self, f):
 		fp = open_file(f)
 		self.init()
-		self.events = filter(lambda a:a, [self.parse(l.strip()) for l in fp])
+		# Filter returns iterator in Py3, list() ensures immediate execution
+		self.events = list(filter(lambda a:a, [self.parse(l.strip()) for l in fp]))
 
 	def structure(self):
 		am = self.appmaster() # this is a copy
@@ -221,7 +222,7 @@ def structure(self):
 		for d in dags:
 			d.structure(vertexes)
 		for a in attempts:
-			if containers.has_key(a.container):
+			if a.container in containers:
 				c = containers[a.container]
 				c.node = a.node
 			else:
@@ -242,7 +243,7 @@ def containers(self):
 		for ev in self.events:
 			if ev.event == "CONTAINER_STOPPED":
 				kvs = csv_kv(ev.args)
-				if containermap.has_key(kvs["containerId"]):
+				if kvs["containerId"] in containermap:
 					containermap[kvs["containerId"]].stop = int(kvs["stoppedTime"])
 					containermap[kvs["containerId"]].status = int(kvs["exitStatus"])
 		return containers
@@ -265,26 +266,31 @@ def attempts(self):
 		key = lambda a:a[0]
 		value = lambda a:a[1]
 		raw = [(csv_kv(ev.args)["taskAttemptId"], ev) for ev in self.events if ev.event == "TASK_ATTEMPT_STARTED" or ev.event == "TASK_ATTEMPT_FINISHED"]
-		pairs = groupby(sorted(raw), key = key)
-		attempts = [Attempt(map(value,p)) for (k,p) in pairs]
+		# FIX: explicitly pass key to sorted() to avoid comparing AMRawEvent objects
+		# which causes TypeError in Python 3
+		pairs = groupby(sorted(raw, key=key), key = key)
+		# Map returns iterator in Py3, list() creates the necessary list
+		attempts = [Attempt(list(map(value,p))) for (k,p) in pairs]
 		return attempts
 
 	def parse(self, l):		
 		if(l.find("[HISTORY]") != -1):
 			m = self.MAIN_RE.match(l)
-			ts = m.group("ts")
-			dag = m.group("dag")
-			event = m.group("event")
-			args = m.group("args")
-			return AMRawEvent(ts, dag, event, args)
+			if m:
+				ts = m.group("ts")
+				dag = m.group("dag")
+				event = m.group("event")
+				args = m.group("args")
+				return AMRawEvent(ts, dag, event, args)
+		return None
 
 def main(argv):
 	tree = AMLog(argv[0]).structure()
 	# AM -> dag -> vertex -> task -> attempt
 	# AM -> container
 	for d in tree.dags:
 		for a in d.attempts():
-			print [a.vertex, a.name, a.container, a.start, a.finish]
+			print([a.vertex, a.name, a.container, a.start, a.finish])
 
 if __name__ == "__main__":
 	main(sys.argv[1:])