Структура папки добавлена на снимок экрана.
Я добавляю код Kafka Producer следующим образом
Код: Выделить всё
import json
import random
import time
from confluent_kafka import SerializingProducer
from datetime import datetime
from faker import Faker
fake = Faker()
def generate_sales():
user = fake.simple_profile()
return {
"transaction_id": fake.uuid4(),
"user_name": user["username"],
"email": fake.email(),
"product_id": random.choice(["product1, product2, product3, product4", "product5", "product6"]),
"product_category": random.choice(["Electronic", "Grocery", "Fashion", "metal works", "sports", "Cosmetics"]),
"price": round(random.uniform(1, 10000), 2),
"quantity": random.randint(1, 1000),
"productBrand": random.choice(["apple", "Nestle", "Tanishq", "Hokkins", "Nike", "Lakma"]),
"currency": random.choice(["INR", "USD", "EUR", "GBP"]),
"transactionDate": datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S'),
"paymentMethod": random.choice(["Payment Gateway", "Bank Transfer", "Cash", "Credit Card", "Debit Card"])
}
def delivery_report(err, msg):
if err is not None:
print(f"Message delivery failed with error {err}")
else:
print(f"Message delivered to topic: {msg.topic} and partition: {msg.partition()}")
def produce_sales_kafka():
topic = 'sales_financial_report'
config = {
'bootstrap.servers': 'localhost:9092'
}
producer = SerializingProducer(
conf=config
)
count = 0
while count < 5:
try:
transactions = generate_sales()
transactions['totalAmount'] = round(transactions['price'] * transactions['quantity'], 2)
print(f'The Total Transaction amount is {transactions["totalAmount"]}')
print("The Json Value",json.dumps(transactions))
producer.produce(
topic=topic,
key=transactions['transaction_id'],
value=json.dumps(transactions),
on_delivery=delivery_report
)
producer.poll(0)
count += 1
time.sleep(1)
except Exception as e:
print('The Error encountered is', e)
# Press the green button in the gutter to run the script.
if __name__ == '__main__':
print("Generating sales transactions...\n")
produce_sales_kafka()
Код: Выделить всё
import os
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, explode
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType
# Define the schema
schema = StructType([
StructField("transaction_id", StringType(), True),
StructField("user_name", StringType(), True),
StructField("email", StringType(), True),
StructField("product_id", StringType(), True),
StructField("product_category", StringType(), True),
StructField("price", FloatType(), True),
StructField("quantity", IntegerType(), True),
StructField("productBrand", StringType(), True),
StructField("currency", StringType(), True),
StructField("transactionDate", StringType(), True),
StructField("paymentMethod", StringType(), True),
StructField("totalAmount", FloatType(), True)
])
def get_spark_context():
spark = SparkSession \
.builder \
.appName("Streaming from Kafka") \
.config("spark.streaming.stopGracefullyOnShutdown", True) \
.config('spark.jars', './jars/kafka-clients-2.6.0.jar,./jars/spark-sql-kafka-0-10_2.12-3.3.0.jar') \
.config("spark.sql.shuffle.partitions", 4) \
.master("local[*]") \
.getOrCreate()
print('Spark session created', spark)
return spark
def connect_kafka():
spark = get_spark_context()
print("-- Connecting to Kafka --")
df = (spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "sales_financial_report")
.option("startingOffsets", "earliest")
.load()
.select(from_json(col('value').cast('string'), schema).alias('sales_transactions'))
.select(col('*'))
)
df.printSchema()
if df is not None:
query = df.writeStream \
.format("csv") \
.option("path", "data/") \
.option("checkpointLocation", "checkpoint/") \
.outputMode("append") \
.queryName("Finance sales") \
.start()
query.awaitTermination()
else:
print("---")
Подробнее здесь: https://stackoverflow.com/questions/786 ... from-kafka
