Works. Need to test out edge cases

docker/A3/T2/kafka-input.csv

@@ -0,0 +1,114 @@
+Andaman and Nicobar,South Andaman,Port Blair,Amaranthus,Amaranthus.1,04/03/2019,6000,8000,7000
+Uttar Pradesh,Agra,Samsabad,Gur(Jaggery),Pathari,04/03/2019,4900.0,5100.0,5000.0
+Telangana,Hyderabad,Mahboob Manison,Dry Chillies,1st Sort,04/03/2019,6000.0,8500.0,7000.0
+Punjab,Mansa,Mansa,Bhindi(Ladies Finger),Bhindi,04/03/2019,3000.0,4000.0,3500.0
+Kerala,Kottayam,Ettumanoor,Cabbage,Other,04/03/2019,2000.0,2200.0,2100.0
+Uttar Pradesh,Agra,Khairagarh,Mustard,Sarson(Black),04/03/2019,3300.0,3500.0,3400.0
+Jammu and Kashmir,Badgam,Zaloosa-Chararishrief (F&V),Cucumbar(Kheera),Cucumbar,04/03/2019,1500.0,1700.0,1600.0
+Punjab,Muktsar,Malout,Spinach,Spinach,04/03/2019,500.0,800.0,600.0
+Gujarat,Vadodara(Baroda),Padra,Tomato,Other,04/03/2019,1200.0,1500.0,1350.0
+Rajasthan,Jalore,Jalore,Lemon,Other,04/03/2019,2300.0,2500.0,2400.0
+Gujarat,Kachchh,Bachau,Guar Seed(Cluster Beans Seed),Whole,04/03/2019,4000.0,4025.0,4010.0
+Uttar Pradesh,Kannuj,Chhibramau(Kannuj),Mustard Oil,Mustard Oil,04/03/2019,8900.0,9090.0,9000.0
+Kerala,Kasargod,Manjeswaram,Little gourd (Kundru),Other,04/03/2019,4200.0,5000.0,4600.0
+Kerala,Kollam,Anchal,Banana,Red Banana,04/03/2019,4700.0,4900.0,4800.0
+Gujarat,Vadodara(Baroda),Padra,Sweet Potato,Other,04/03/2019,1500.0,2000.0,1700.0
+Gujarat,Panchmahals,Gogamba,Paddy(Dhan)(Common),G. R. 11,04/03/2019,1500.0,1550.0,1525.0
+Uttar Pradesh,Aligarh,Atrauli,Maize,Hybrid,04/03/2019,1930.0,1950.0,1940.0
+Odisha,Mayurbhanja,Betnoti,Leafy Vegetable,Other,04/03/2019,1400.0,1600.0,1500.0
+Tamil Nadu,Coimbatore,Annur,Tobacco,Other,04/03/2019,3000.0,3200.0,3100.0
+Uttar Pradesh,Bhadohi(Sant Ravi Nagar),Gopiganj,Potato,Potato,04/03/2019,350.0,400.0,400.0
+Maharashtra,Pune,Pune,Cucumbar(Kheera),Other,04/03/2019,1000.0,3000.0,2000.0
+Tamil Nadu,Coimbatore,Palladam,Coconut,Other,04/03/2019,2040.0,2240.0,2140.0
+West Bengal,Burdwan,Burdwan,Sweet Pumpkin,Sweet Pumpkin,04/03/2019,950.0,1050.0,1000.0
+Manipur,Bishnupur,Bishenpur,Paddy(Dhan)(Common),Other,04/03/2019,2250.0,2250.0,2250.0
+Uttar Pradesh,Saharanpur,Sultanpurchilkana,Carrot,Carrot,04/03/2019,400.0,500.0,450.0
+Kerala,Thirssur,Kodungalloor,Tomato,Other,04/03/2019,1600.0,1600.0,1600.0
+Assam,Kamrup,Pamohi(Garchuk),Pumpkin,Pumpkin,04/03/2019,800.0,1000.0,900.0
+Uttar Pradesh,Agra,Samsabad,Barley (Jau),Dara,04/03/2019,1825.0,1875.0,1850.0
+Punjab,Muktsar,Malout,Green Chilli,Green Chilly,04/03/2019,800.0,3500.0,2000.0
+Uttar Pradesh,Muzaffarnagar,Muzzafarnagar,Gur(Jaggery),Yellow,04/03/2019,2400.0,2600.0,2510.0
+Punjab,Hoshiarpur,Dasuya,Kinnow,Other,04/03/2019,1600.0,2000.0,2000.0
+Rajasthan,Jalore,Jalore,Pomegranate,Other,04/03/2019,1400.0,1600.0,1500.0
+Uttar Pradesh,Baghpat,Bagpat,Carrot,Carrot,04/03/2019,400.0,600.0,500.0
+West Bengal,Bankura,Bishnupur(Bankura),Rice,Fine,04/03/2019,2800.0,2900.0,2850.0
+Kerala,Alappuzha,Mannar,Carrot,Other,04/03/2019,3700.0,3900.0,3800.0
+Odisha,Sundergarh,Bonai,Potato,Other,04/03/2019,1000.0,1000.0,1000.0
+Tamil Nadu,Villupuram,Manalurpet,Ragi (Finger Millet),Other,04/03/2019,2312.0,2312.0,2312.0
+Uttar Pradesh,Bahraich,Ruperdeeha,Grapes,Green,04/03/2019,4480.0,4540.0,4500.0
+Maharashtra,Satara,Karad,Cabbage,Other,04/03/2019,500.0,1000.0,1000.0
+Tamil Nadu,Cuddalore,Cuddalore,Paddy(Dhan)(Common),B P T,04/03/2019,1344.0,1555.0,1500.0
+Kerala,Kottayam,Ettumanoor,Onion,Big,04/03/2019,3300.0,3500.0,3500.0
+West Bengal,Medinipur(W),Ghatal,"Sesamum(Sesame,Gingelly,Til)",Red,04/03/2019,8500.0,8700.0,8600.0
+Rajasthan,Jalore,Jalore,Bottle gourd,Other,04/03/2019,1000.0,1200.0,1100.0
+Uttar Pradesh,Baghpat,Baraut,Raddish,Raddish,04/03/2019,400.0,600.0,500.0
+Maharashtra,Buldhana,Deoulgaon Raja,Cotton,Desi,04/03/2019,5100.0,5400.0,5300.0
+Kerala,Thirssur,Chelakkara,Elephant Yam (Suran),Other,04/03/2019,2500.0,2800.0,2600.0
+Tamil Nadu,Salem,Salem,Turmeric,Finger,04/03/2019,6940.0,7395.0,7120.0
+Odisha,Ganjam,Digapahandi,Maize,Other,04/03/2019,1300.0,1420.0,1350.0
+Gujarat,Vadodara(Baroda),Padra,Brinjal,Other,04/03/2019,400.0,1000.0,750.0
+Uttar Pradesh,Agra,Achnera,Barley (Jau),Dara,04/03/2019,1800.0,1860.0,1830.0
+Uttar Pradesh,Muzaffarnagar,Muzzafarnagar,Green Chilli,Green Chilly,04/03/2019,2230.0,2300.0,2260.0
+Haryana,Gurgaon,Pataudi,Spinach,Other,04/03/2019,500.0,1000.0,750.0
+Kerala,Malappuram,Kondotty,Tomato,Other,04/03/2019,1000.0,1200.0,1100.0
+Odisha,Balasore,Jaleswar,Onion,Other,04/03/2019,1000.0,1150.0,1100.0
+Tamil Nadu,Villupuram,Vikkiravandi,Hybrid Cumbu,Other,04/03/2019,1800.0,1853.0,1825.0
+Kerala,Malappuram,Kondotty,Cowpea(Veg),Other,04/03/2019,3000.0,3200.0,3100.0
+Kerala,Malappuram,Kondotty,Lemon,Other,04/03/2019,4000.0,4200.0,4100.0
+Punjab,Tarntaran,Patti,Spinach,Other,04/03/2019,300.0,430.0,400.0
+Uttar Pradesh,Saharanpur,Nanuta,Green Chilli,Green Chilly,04/03/2019,1200.0,1300.0,1250.0
+Manipur,Imphal East,Lamlong Bazaar,Onion,Other,04/03/2019,1300.0,2500.0,2500.0
+Tripura,North Tripura,Dasda,Rice,Masuri,04/03/2019,2500.0,2700.0,2600.0
+Rajasthan,Kota,Itawa,Wheat,Other,04/03/2019,1915.0,1915.0,1915.0
+Uttar Pradesh,Sonbhadra,Robertsganj,Garlic,Garlic,04/03/2019,1750.0,1840.0,1810.0
+Odisha,Dhenkanal,Hindol,Brinjal,Brinjal,04/03/2019,800.0,1200.0,1000.0
+Uttar Pradesh,Bulandshahar,Divai,Pumpkin,Pumpkin,04/03/2019,680.0,700.0,690.0
+Meghalaya,East Khasi Hills,Shillong,Ginger(Dry),Dry,04/03/2019,6000.0,7000.0,6500.0
+Uttar Pradesh,Kannuj,Chhibramau(Kannuj),Cauliflower,Local,04/03/2019,450.0,500.0,480.0
+Madhya Pradesh,Dindori,Gorakhpur,Paddy(Dhan)(Common),Other,04/03/2019,1300.0,1350.0,1325.0
+Haryana,Mewat,Taura,Potato,Potato,04/03/2019,700.0,700.0,700.0
+Manipur,Bishnupur,Bishenpur,Cabbage,Cabbage,04/03/2019,1100.0,1300.0,1300.0
+Uttar Pradesh,Muzaffarnagar,Muzzafarnagar,Pumpkin,Pumpkin,04/03/2019,750.0,830.0,800.0
+Uttar Pradesh,Kannuj,Chhibramau(Kannuj),Mustard,Lohi Black,04/03/2019,3300.0,3450.0,3400.0
+Maharashtra,Satara,Karad,Carrot,Other,04/03/2019,500.0,1000.0,1000.0
+Uttar Pradesh,Saharanpur,Gangoh,Tomato,Deshi,04/03/2019,1325.0,1350.0,1340.0
+Kerala,Kasargod,Manjeswaram,Bottle gourd,Bottle Gourd,04/03/2019,1600.0,2000.0,1800.0
+Uttar Pradesh,Muzaffarnagar,Thanabhawan,Cauliflower,Local,04/03/2019,300.0,400.0,340.0
+Uttar Pradesh,Agra,Jarar,Green Chilli,Green Chilly,04/03/2019,1040.0,1120.0,1080.0
+Himachal Pradesh,Kangra,Kangra(Jassour),Methi(Leaves),Other,04/03/2019,900.0,1000.0,950.0
+Maharashtra,Satara,Karad,Spinach,Other,04/03/2019,1.0,2.0,2.0
+Gujarat,Bharuch,Ankleshwar,Methi(Leaves),Methi,04/03/2019,600.0,1000.0,825.0
+Kerala,Thiruvananthapuram,Maranelloor,Banana,Poovan,04/03/2019,2900.0,3200.0,3000.0
+Tamil Nadu,Villupuram,Villupuram,Black Gram (Urd Beans)(Whole),Other,04/03/2019,2559.0,2639.0,2619.0
+West Bengal,Puruliya,Balarampur,Tomato,Other,04/03/2019,250.0,350.0,300.0
+Kerala,Thiruvananthapuram,Parassala,Snakeguard,Snakeguard,04/03/2019,2600.0,3000.0,2600.0
+Kerala,Thirssur,Irinjalakkuda,Bitter gourd,Other,04/03/2019,4000.0,4000.0,4000.0
+Uttar Pradesh,Ghaziabad,Hapur,Wheat,Dara,04/03/2019,1900.0,2100.0,1980.0
+Punjab,Jalandhar,Mehatpur,Spinach,Other,04/03/2019,400.0,600.0,600.0
+Himachal Pradesh,Kangra,Kangra(Jassour),Pomegranate,Other,04/03/2019,4000.0,4500.0,4250.0
+Assam,Kamrup,Pamohi(Garchuk),Brinjal,Brinjal,04/03/2019,1200.0,1400.0,1300.0
+Punjab,Muktsar,Malout,Peas cod,Other,04/03/2019,1000.0,1500.0,1200.0
+Kerala,Thiruvananthapuram,Parassala,Bitter gourd,Bitter Gourd,04/03/2019,5000.0,5500.0,5000.0
+Rajasthan,Jalore,Jalore,Bhindi(Ladies Finger),Other,04/03/2019,3800.0,4000.0,3900.0
+Kerala,Thirssur,Irinjalakkuda,Elephant Yam (Suran),Other,04/03/2019,2400.0,2400.0,2400.0
+Uttar Pradesh,Agra,Fatehabad,Barley (Jau),Dara,04/03/2019,1700.0,1750.0,1720.0
+Kerala,Alappuzha,Chengannur,Ashgourd,Other,04/03/2019,1100.0,1400.0,1200.0
+Tamil Nadu,Villupuram,Tindivanam,Groundnut,Other,04/03/2019,6236.0,7387.0,6724.0
+Kerala,Malappuram,Kondotty,Potato,Other,04/03/2019,2000.0,2200.0,2100.0
+Uttar Pradesh,Baghpat,Bagpat,Tomato,Deshi,04/03/2019,1100.0,1300.0,1200.0
+Tripura,South District,Barpathari,Potato,Jalander,04/03/2019,700.0,800.0,750.0
+West Bengal,Puruliya,Purulia,Moath Dal,Moath Dal,04/03/2019,7300.0,7400.0,7400.0
+Maharashtra,Kolhapur,Kolhapur,Brinjal,Other,04/03/2019,600.0,3000.0,1800.0
+Punjab,Ludhiana,Sahnewal,Methi(Leaves),Methi,04/03/2019,1000.0,1000.0,1000.0
+Punjab,Muktsar,Malout,Guava,Guava,04/03/2019,2000.0,2500.0,2200.0
+Tamil Nadu,Erode,Dharapuram,Coconut,Big,04/03/2019,2505.0,3275.0,3000.0
+Maharashtra,Nagpur,Savner,Wheat,Other,04/03/2019,1935.0,2040.0,2000.0
+Punjab,Jalandhar,Adampur,Amphophalus,Other,04/03/2019,1800.0,2000.0,2000.0
+Uttar Pradesh,Lakhimpur,Paliakala,Bottle gourd,Bottle Gourd,04/03/2019,1020.0,1080.0,1050.0
+Uttar Pradesh,Muzaffarnagar,Thanabhawan,Brinjal,Brinjal,04/03/2019,400.0,500.0,440.0
+West Bengal,Nadia,Nadia,Mustard,Yellow (Black),04/03/2019,4000.0,4150.0,4100.0
+Gujarat,Bharuch,Ankleshwar,Green Chilli,Green Chilly,04/03/2019,2600.0,3300.0,3000.0
+Haryana,Ambala,Shahzadpur,Ber(Zizyphus/Borehannu),Ber(Zizyphus),04/03/2019,2000.0,4000.0,3000.0
+Uttar Pradesh,Agra,Fatehabad,Onion,Red,04/03/2019,850.0,1000.0,900.0
+Tamil Nadu,Namakkal,Namagiripettai,Turmeric,Bulb,04/03/2019,5050.0,6050.0,5740.0
+EOF

docker/hadoop3-A3.dockerfile

@@ -79,6 +79,8 @@ RUN mv page_embeddings.json /A2/
 RUN mv w /A2/
 RUN mv adjacency_list.txt /A2/
 
+ADD A3 /A3
+
 # add new ports here
 EXPOSE 8088 50070 50075 50030 50060 9870 10000 19888
 # RUN swapoff -a

docker/hadoop_server-A3.py

@@ -31,7 +31,7 @@
 
 JOBHISTORY_URL = "http://localhost:19888/ws/v1/history/mapreduce/jobs"
 
-TASK_OUTPUT_PATH = {"A3T1":"task-1-output", "A3T2":"task-2-output"}
+TASK_OUTPUT_PATH = {"A3T1":"task-1-output", "A3T2":"task-2-output", "A2T1":"task-1-output", "A2T2":"task-2-output"}
 
 FILEPATH = os.path.join(os.getcwd(), 'output')
 
@@ -208,6 +208,15 @@ def run_spark_job(team_id, assignment_id, submission_id, timeout, spark:str):
     st = os.stat(os.path.join(task_path,"spark.py"))
     os.chmod(os.path.join(task_path,"spark.py"), st.st_mode | stat.S_IEXEC)
 
+    res = create_hdfs_directory(f"/{team_id}")
+    if res != 0:
+        print(f"Failed to create HDFS Directory : hdfs:/{team_id}")
+
+    directory = f"/{team_id}/{assignment_id}"
+    res = create_hdfs_directory(directory)
+    if res != 0:
+        print(f"Failed to create HDFS Directory : hdfs:{directory}")
+
     task_path = os.path.join(path, submission_id)
 
     timestamp = str(time.time())
@@ -221,7 +230,7 @@ def run_spark_job(team_id, assignment_id, submission_id, timeout, spark:str):
     if assignment_id == "A3T1":
         logger.mark(f"Spawning Spark Process")
 
-        spark_job = f'''{SPARK} --name={job_name} --master=yarn --deploy-mode=cluster "/{os.path.join(task_path,'spark.py')}" "/A3/spark-input.csv" "/{os.path.join(task_path,'spark-output')}"'''
+        spark_job = f'''{SPARK} --name={job_name} --master=yarn --deploy-mode=cluster /{os.path.join(task_path,'spark.py')} /A3/T1/spark-input.csv /{team_id}/{assignment_id}/{TASK_OUTPUT_PATH[assignment_id]}'''
 
         spark_process = subprocess.Popen([
             spark_job
@@ -240,7 +249,7 @@ def run_spark_job(team_id, assignment_id, submission_id, timeout, spark:str):
             flag = False
         else:
             current_job = data[0]
-            current_job_state = current_job["state"]
+            current_job_state = current_job["finalStatus"]
             application_id = current_job["id"]
 
         if not os.path.exists(os.path.join(FILEPATH, team_id, assignment_id)):
@@ -252,12 +261,12 @@ def run_spark_job(team_id, assignment_id, submission_id, timeout, spark:str):
             job_output = "Good Job!"
             status = current_job["state"]
 
-            if os.path.exists(os.path.join(FILEPATH, team_id, assignment_id, "part-*")):
-                os.remove(os.path.join(FILEPATH, team_id, assignment_id, "part-*"))
+            if os.path.exists(os.path.join(FILEPATH, team_id, assignment_id, "part-00000")):
+                os.remove(os.path.join(FILEPATH, team_id, assignment_id, "part-00000"))
                 # os.remove(os.path.join(FILEPATH, team_id, assignment_id, "_SUCCESS"))
-            
-            process = subprocess.Popen([f"cp -R /{os.join(task_path, 'spark-output', 'part-*')} /{os.path.join(FILEPATH, team_id, assignment_id)}"], shell=True, text=True)
-            _ = process.wait()
+
+            process = subprocess.Popen([f"{HDFS} dfs -get /{team_id}/{assignment_id}/{TASK_OUTPUT_PATH[assignment_id]}/part-* {os.path.join(FILEPATH, team_id, assignment_id)}"], shell=True, text=True)
+            process_code = process.wait()
 
             process = subprocess.Popen([f"mv /{os.path.join(FILEPATH, team_id, assignment_id, 'part-*')} /{os.path.join(FILEPATH, team_id, assignment_id, 'part-00000')}"], shell=True, text=True)
             _ = process.wait()
@@ -362,41 +371,85 @@ def run_kafka_job(team_id, assignment_id, submission_id, timeout, producer:str,
         if os.path.exists(os.path.join(FILEPATH, team_id, assignment_id, "output.json")):
                 os.remove(os.path.join(FILEPATH, team_id, assignment_id, "output.json"))
 
-        consumer_cmd = f'''python3 /{os.path.join(task_path, "consumer.py")} agriculture > /{os.path.join(FILEPATH, team_id, assignment_id, "output.json")}'''
-        producer_cmd = f'''cat /A3/dataset.csv | python3 /{os.path.join(task_path, "producer.py")} agriculture'''
+        consumer_cmd = f'''python3 /{os.path.join(task_path, "consumer.py")} "agriculture" > /{os.path.join(FILEPATH, team_id, assignment_id, "output.json")}'''
+        producer_cmd = f'''cat /A3/T2/kafka-input.csv | python3 /{os.path.join(task_path, "producer.py")} "agriculture"'''
 
+        print("Starting Consumer Process")
         consumer_process = subprocess.Popen([
             consumer_cmd
         ], shell=True, text=True, preexec_fn=os.setsid, stdout=subprocess.PIPE)
 
+        time.sleep(5)
+
+        print("Starting Producer Process")
         producer_process = subprocess.Popen([
             producer_cmd
         ], shell=True, text=True, preexec_fn=os.setsid, stdout=subprocess.PIPE)
 
-        kafka_process_code1 = producer_process.wait(timeout=timeout)
-        kafka_process_code2 = consumer_process.wait(timeout=timeout)
+        flag = True
 
-        if os.path.exists(os.path.join(FILEPATH, team_id, assignment_id, "output.json")):
+        error_logs = [
+                    f"Submission ID : {submission_id}",
+                    f"Team ID : {team_id}",
+                    f"Assignment ID : {assignment_id}",
+                    "Note : If you do not see any error with respect to your code and you only see : \n\nlog4j:WARN\n\nThen that means your code had infinite loop and submission was killed.\n\nLOGS :- \n\n",
+                ]
+
+        consumer_stderr = consumer_process.communicate()[1]
+        if consumer_stderr:
+            error_logs.append(
+                    "consumer\n--------\n\n"+consumer_stderr
+            )
+            consumer_process.kill()
+
+        try:
+            kafka_process_code2 = consumer_process.wait(timeout=(timeout-10)//2)
+            print("Consumer Completed!")
+        except subprocess.TimeoutExpired:
+            logger.mark(f"Team ID : {team_id} Assignment ID : {assignment_id} Kafka Job Failed")
+            msg = f"Team ID : {team_id} Assignment ID : {assignment_id} Kafka Job Failed. Time Limit Exceeded in Consumer."
+            status = "FAILED"
+            job_output = "Failed! Time Limit Exceeded in Consumer."
+            flag = False
+
+        producer_stderr = producer_process.communicate()[1]
+        if producer_stderr:
+            error_logs.append(
+                    "producer\n--------\n\n"+producer_stderr
+            )
+            producer_process.kill()
+
+        try:
+            kafka_process_code1 = producer_process.wait(timeout=(timeout-10)//2)
+            print("Producer Completed!")
+        except subprocess.TimeoutExpired:
+            logger.mark(f"Team ID : {team_id} Assignment ID : {assignment_id} Kafka Job Failed")
+            msg = f"Team ID : {team_id} Assignment ID : {assignment_id} Kafka Job Failed. Time Limit Exceeded in Producer."
+            status = "FAILED"
+            job_output = "Failed! Time Limit Exceeded in Producer."
+            flag = False
+
+        if flag and os.path.exists(os.path.join(FILEPATH, team_id, assignment_id, "output.json")):
             if os.stat(os.path.join(FILEPATH, team_id, assignment_id, "output.json")).st_size == 0 or (kafka_process_code1 != 0 or kafka_process_code2 != 0):
                 logger.mark(f"Team ID : {team_id} Assignment ID : {assignment_id} Kafka Job Failed")
                 msg = f"Team ID : {team_id} Assignment ID : {assignment_id} Kafka Job Failed."
                 status = "FAILED"
                 job_output = "Failed! Your submission might have thrown error(s). Logs have been mailed to you."
 
-                error_logs = [
-                    f"Submission ID : {submission_id}",
-                    f"Team ID : {team_id}",
-                    f"Assignment ID : {assignment_id}",
-                    "Note : If you do not see any error with respect to your code and you only see : \n\nlog4j:WARN\n\nThen that means your code had infinite loop and submission was killed.\n\nLOGS :- \n\n",
-                ]
+                # error_logs = [
+                #     f"Submission ID : {submission_id}",
+                #     f"Team ID : {team_id}",
+                #     f"Assignment ID : {assignment_id}",
+                #     "Note : If you do not see any error with respect to your code and you only see : \n\nlog4j:WARN\n\nThen that means your code had infinite loop and submission was killed.\n\nLOGS :- \n\n",
+                # ]
 
-                error_logs.append(
-                    "producer\n--------\n\n"+producer_process.communicate()[1]
-                )
+                # error_logs.append(
+                #     "producer\n--------\n\n"+producer_process.communicate()[1]
+                # )
 
-                error_logs.append(
-                    "consumer\n--------\n\n"+consumer_process.communicate()[1]
-                )
+                # error_logs.append(
+                #     "consumer\n--------\n\n"+consumer_process.communicate()[1]
+                # )
 
                 with open(os.path.join(FILEPATH, team_id, assignment_id, "error.txt"), "w+") as f:
                     f.write(error_logs)
@@ -800,11 +853,17 @@ def initialize_environment(add_dataset=True) -> Tuple:
         _ = create_hdfs_directory("/A2")
         _ = create_hdfs_directory("/A2/input")
 
+        _ = create_hdfs_directory("/A3")
+        _ = create_hdfs_directory("/A3/T1")
+
         p = subprocess.Popen([f"{HDFS} dfs -put /A2/graph.txt /A2/input"], shell=True, text=True)
         _ = p.wait()
 
         p = subprocess.Popen([f"{HDFS} dfs -put /A2/adjacency_list.txt /A2/input"], shell=True, text=True)
         _ = p.wait()
+
+        p = subprocess.Popen([f"{HDFS} dfs -put /A3/T1/spark-input.csv /A3/T1/"], shell=True, text=True)
+        _ = p.wait()
 
     return 0, error_logs 
 

flask_backend/app.py

@@ -319,7 +319,7 @@ def sanity_check():
                 consumer.write(consumer_data)
                 consumer.close()                
 
-        process = subprocess.Popen([f'pylint --disable=I,R,C,W {os.path.join(os.getcwd(), "compile-test/")}'], shell=True, stdout=subprocess.PIPE, text=True)
+        process = subprocess.Popen([f'pylint --disable=R,C,W,import-error {os.path.join(os.getcwd(), "compile-test/")}'], shell=True, stdout=subprocess.PIPE, text=True)
         exit_code = process.wait()
 
         output = process.communicate()[0]

job_tracker/executor.py

@@ -313,13 +313,13 @@ def cleanup(self):
         fetch_ip=BACKEND_INTERNAL_IP,
         fetch_port=9000,
         fetch_route="get-jobs",
-        num_workers=4,
+        num_workers=1,
         global_queue_thread=True,
         global_prefetch_thread=True,
         prefetch_threads=4,
         prefetch_factor=4,
         threshold=10,
-        num_backends=8
+        num_backends=1
     )
 
     print(executor)
@@ -329,8 +329,8 @@ def cleanup(self):
     docker_route = "run_job"
     docker_image = "hadoop-3.2.2:0.1"
 
-    backend_cpu_limit: int = 3
-    backend_mem_limit: str = "12000m"
+    backend_cpu_limit: int = 6
+    backend_mem_limit: str = "8000m"
     backend_host_output_dir: str = f"{os.path.join(os.getcwd(),'output')}"
     backend_docker_output_dir: str = f"/output"
     backend_memswapiness: int = 0