black mamba

Author: huseinzol05

apache-bigdata/1.flask-hadoop/app.py

@@ -9,35 +9,62 @@
 
 print(hdfs.hdfs().list_directory('/user'))
 
+
 @app.route('/')
 def hello_world():
     return 'Hey, we have Flask in a Docker container!'
 
+
 @app.route('/test', methods = ['POST'])
 def test():
     f = request.files['file']
-    f.save(os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(f.filename)))
+    f.save(
+        os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(f.filename))
+    )
     return f.filename
 
+
 @app.route('/wordcount', methods = ['POST'])
 def wordcount():
     f = request.files['file']
-    f.save(os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(f.filename)))
-    with open(f.filename,'r') as fopen:
+    f.save(
+        os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(f.filename))
+    )
+    with open(f.filename, 'r') as fopen:
         hdfs.dump(fopen.read(), '/user/input_wordcount/text')
-    os.system('pydoop script -c combiner wordcount.py /user/input_wordcount /user/output_wordcount')
+    os.system(
+        'pydoop script -c combiner wordcount.py /user/input_wordcount /user/output_wordcount'
+    )
     list_files = hdfs.hdfs().list_directory('/user/output_wordcount')
-    return json.dumps([hdfs.load(file['name'], mode='rt') for file in list_files if 'SUCCESS' not in file['name']])
+    return json.dumps(
+        [
+            hdfs.load(file['name'], mode = 'rt')
+            for file in list_files
+            if 'SUCCESS' not in file['name']
+        ]
+    )
+
 
 @app.route('/lowercase', methods = ['POST'])
 def lowercase():
     f = request.files['file']
-    f.save(os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(f.filename)))
-    with open(f.filename,'r') as fopen:
+    f.save(
+        os.path.join(app.config['UPLOAD_FOLDER'], secure_filename(f.filename))
+    )
+    with open(f.filename, 'r') as fopen:
         hdfs.dump(fopen.read(), '/user/input_lowercase/text')
-    os.system("pydoop script --num-reducers 0 -t '' lowercase.py /user/input_lowercase /user/output_lowercase")
+    os.system(
+        "pydoop script --num-reducers 0 -t '' lowercase.py /user/input_lowercase /user/output_lowercase"
+    )
     list_files = hdfs.hdfs().list_directory('/user/output_lowercase')
-    return json.dumps([hdfs.load(file['name'], mode='rt') for file in list_files if 'SUCCESS' not in file['name']])
+    return json.dumps(
+        [
+            hdfs.load(file['name'], mode = 'rt')
+            for file in list_files
+            if 'SUCCESS' not in file['name']
+        ]
+    )
+
 
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0',port=5000)
+    app.run(debug = True, host = '0.0.0.0', port = 5000)

apache-bigdata/1.flask-hadoop/lowercase.py

@@ -1,2 +1,2 @@
 def mapper(_, record, writer):
-    writer.emit("", record.lower())
+    writer.emit('', record.lower())

apache-bigdata/1.flask-hadoop/wordcount.py

@@ -1,10 +1,12 @@
 def mapper(_, text, writer):
     for word in text.split():
-        writer.emit(word, "1")
+        writer.emit(word, '1')
+
 
 def reducer(word, icounts, writer):
     writer.emit(word, sum(map(int, icounts)))
 
+
 def combiner(word, icounts, writer):
     writer.count('combiner calls', 1)
     reducer(word, icounts, writer)

apache-bigdata/2.flask-kafka/app.py

@@ -2,18 +2,28 @@
 from kafka import KafkaConsumer
 import json
 import os
+
 app = Flask(__name__)
 
+
 @app.route('/')
 def hello_world():
     return 'Hey, we have Flask in a Docker container!'
 
-@app.route("/topic/<string:topic>/")
+
+@app.route('/topic/<string:topic>/')
 def get_topics(topic):
-    consumer = KafkaConsumer(topic, auto_offset_reset='earliest',
-                             bootstrap_servers=['localhost:9092'],
-                             api_version=(0, 10), consumer_timeout_ms=1000)
-    return json.dumps([json.loads(msg.value.decode('utf-8')) for msg in consumer])
+    consumer = KafkaConsumer(
+        topic,
+        auto_offset_reset = 'earliest',
+        bootstrap_servers = ['localhost:9092'],
+        api_version = (0, 10),
+        consumer_timeout_ms = 1000,
+    )
+    return json.dumps(
+        [json.loads(msg.value.decode('utf-8')) for msg in consumer]
+    )
+
 
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0',port=5000)
+    app.run(debug = True, host = '0.0.0.0', port = 5000)

apache-bigdata/2.flask-kafka/producer.py

@@ -4,6 +4,7 @@
 from bs4 import BeautifulSoup
 from kafka import KafkaProducer
 
+
 def parse(markup):
     title = '-'
     submit_by = '-'
@@ -24,7 +25,10 @@ def parse(markup):
         if ingredients_section:
             for ingredient in ingredients_section:
                 ingredient_text = ingredient.text.strip()
-                if 'Add all ingredients to list' not in ingredient_text and ingredient_text != '':
+                if (
+                    'Add all ingredients to list' not in ingredient_text
+                    and ingredient_text != ''
+                ):
                     ingredients.append({'step': ingredient.text.strip()})
 
         if description_section:
@@ -36,20 +40,26 @@ def parse(markup):
         if title_section:
             title = title_section[0].text
 
-        rec = {'title': title, 'submitter': submit_by, 'description': description, 'calories': calories,
-               'ingredients': ingredients}
+        rec = {
+            'title': title,
+            'submitter': submit_by,
+            'description': description,
+            'calories': calories,
+            'ingredients': ingredients,
+        }
 
     except Exception as ex:
         print('Exception while parsing')
         print(str(ex))
     finally:
         return json.dumps(rec)
 
+
 def publish_message(producer_instance, topic_name, key, value):
     try:
-        key_bytes = bytes(key, encoding='utf-8')
-        value_bytes = bytes(value, encoding='utf-8')
-        producer_instance.send(topic_name, key=key_bytes, value=value_bytes)
+        key_bytes = bytes(key, encoding = 'utf-8')
+        value_bytes = bytes(value, encoding = 'utf-8')
+        producer_instance.send(topic_name, key = key_bytes, value = value_bytes)
         producer_instance.flush()
         print('Message published successfully.')
     except Exception as ex:
@@ -60,7 +70,9 @@ def publish_message(producer_instance, topic_name, key, value):
 def connect_kafka_producer():
     _producer = None
     try:
-        _producer = KafkaProducer(bootstrap_servers=['localhost:9092'], api_version=(0, 10))
+        _producer = KafkaProducer(
+            bootstrap_servers = ['localhost:9092'], api_version = (0, 10)
+        )
     except Exception as ex:
         print('Exception while connecting Kafka')
         print(str(ex))
@@ -72,7 +84,7 @@ def fetch_raw(recipe_url):
     html = None
     print('Processing..{}'.format(recipe_url))
     try:
-        r = requests.get(recipe_url, headers=headers)
+        r = requests.get(recipe_url, headers = headers)
         if r.status_code == 200:
             html = r.text
     except Exception as ex:
@@ -89,7 +101,7 @@ def get_recipes():
     print('Accessing list')
 
     try:
-        r = requests.get(url, headers=headers)
+        r = requests.get(url, headers = headers)
         if r.status_code == 200:
             html = r.text
             soup = BeautifulSoup(html, 'lxml')
@@ -110,7 +122,7 @@ def get_recipes():
 if __name__ == '__main__':
     headers = {
         'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/66.0.3359.181 Safari/537.36',
-        'Pragma': 'no-cache'
+        'Pragma': 'no-cache',
     }
 
     all_recipes = get_recipes()

apache-bigdata/3.flask-hadoop-hive/app.py

@@ -4,19 +4,26 @@
 from pyhive import hive
 
 cursor = hive.connect('localhost').cursor()
-cursor.execute("CREATE TABLE IF NOT EXISTS employee ( eid int, name String, salary String, destignation String) COMMENT 'employee details' ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' STORED AS TEXTFILE")
-cursor.execute("LOAD DATA LOCAL INPATH 'sample.txt' OVERWRITE INTO TABLE employee")
+cursor.execute(
+    "CREATE TABLE IF NOT EXISTS employee ( eid int, name String, salary String, destignation String) COMMENT 'employee details' ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' STORED AS TEXTFILE"
+)
+cursor.execute(
+    "LOAD DATA LOCAL INPATH 'sample.txt' OVERWRITE INTO TABLE employee"
+)
 
 app = Flask(__name__)
 
+
 @app.route('/')
 def hello_world():
     return 'Hey, we have Flask in a Docker container!'
 
+
 @app.route('/employee/<string:person>/')
 def get(person):
-    cursor.execute("SELECT * FROM employee WHERE name = '%s'"%(person))
+    cursor.execute("SELECT * FROM employee WHERE name = '%s'" % (person))
     return json.dumps(list(cursor.fetchall()))
 
+
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0',port=5000)
+    app.run(debug = True, host = '0.0.0.0', port = 5000)

basic-backend/1.flask-hello/app.py

@@ -1,17 +1,22 @@
 from flask import Flask
+
 app = Flask(__name__)
 
+
 @app.route('/')
 def hello_world():
     return 'Hey, we have Flask in a Docker container!'
 
-@app.route("/members")
+
+@app.route('/members')
 def members():
-    return "you can put anything after /members/"
- 
-@app.route("/members/<string:name>/")
+    return 'you can put anything after /members/'
+
+
+@app.route('/members/<string:name>/')
 def getMember(name):
     return name
 
+
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0',port=5000)
+    app.run(debug = True, host = '0.0.0.0', port = 5000)

basic-backend/2.flask-mongodb/app.py

@@ -1,23 +1,27 @@
 from flask import Flask, request
 from pymongo import MongoClient
 import json
+
 app = Flask(__name__)
 client = MongoClient('localhost', 27017)
 db = client.test_database
 posts = db.posts
 
+
 @app.route('/')
 def hello_world():
     return 'Hey, we have Flask with MongoDB in a Docker container!'
 
-@app.route('/insert',methods = ['GET'])
+
+@app.route('/insert', methods = ['GET'])
 def insert():
     if not request.args.get('name'):
         return 'insert name'
-    posts.insert_one({'name':request.args.get('name')})
-    return 'done inserted '+ request.args.get('name')
+    posts.insert_one({'name': request.args.get('name')})
+    return 'done inserted ' + request.args.get('name')
 
-@app.route('/get',methods = ['GET'])
+
+@app.route('/get', methods = ['GET'])
 def get():
     if not request.args.get('name'):
         return 'insert name'
@@ -26,5 +30,6 @@ def get():
     except:
         return 'not found'
 
+
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0',port=5000)
+    app.run(debug = True, host = '0.0.0.0', port = 5000)

basic-backend/3.flask-rest-api/app.py

@@ -1,5 +1,6 @@
 from flask import Flask, request
 from flask_restful import reqparse, abort, Api, Resource
+
 app = Flask(__name__)
 api = Api(app)
 
@@ -11,28 +12,33 @@
     'todo3': {'task': 'profit!'},
 }
 
+
 def abort_if_todo_doesnt_exist(todo_id):
     if todo_id not in TODOS:
-        abort(404, message="Todo {} doesn't exist".format(todo_id))
+        abort(404, message = "Todo {} doesn't exist".format(todo_id))
+
 
 parser = reqparse.RequestParser()
 parser.add_argument('task')
 
+
 class HelloWorld(Resource):
     def get(self):
         return {'hello': 'world'}
 
+
 class TodoSimple(Resource):
     def get(self, todo_id):
         try:
             return {todo_id: todos[todo_id]}
         except:
-            return {'error':'todo not found'}
+            return {'error': 'todo not found'}
 
     def put(self, todo_id):
         todos[todo_id] = request.form['data']
         return {todo_id: todos[todo_id]}
 
+
 class Todo(Resource):
     def get(self, todo_id):
         abort_if_todo_doesnt_exist(todo_id)
@@ -49,6 +55,7 @@ def put(self, todo_id):
         TODOS[todo_id] = task
         return task, 201
 
+
 class TodoList(Resource):
     def get(self):
         return TODOS
@@ -60,10 +67,11 @@ def post(self):
         TODOS[todo_id] = {'task': args['task']}
         return TODOS[todo_id], 201
 
+
 api.add_resource(HelloWorld, '/')
 api.add_resource(TodoSimple, '/<string:todo_id>')
 api.add_resource(Todo, '/todos/<todo_id>')
 api.add_resource(TodoList, '/todos')
 
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0',port=5000)
+    app.run(debug = True, host = '0.0.0.0', port = 5000)