frompyspark.sqlimportSparkSession, functionsasFfrompyspark.sql.typesimportStructType, StructField, StringType, ArrayTypeclassProfilePipeline:
def__init__(
self,
name="Mariana Costa",
role="data and more stuff",
education="B.E. Production Engineering - UFRN (Brazil)",
interests=("music", "HQ", "data literacy"),
stack=("PySpark", "Databricks", "Terraform", "SQL", "Python"),
app_name="mrncstt-profile",
):
self.name=nameself.role=roleself.education=educationself.interests=list(interests)
self.stack=list(stack)
self.spark=SparkSession.builder.appName(app_name).getOrCreate()
defbuild_profile(self):
schema=StructType([
StructField("name", StringType()),
StructField("role", StringType()),
StructField("education", StringType()),
StructField("interests", ArrayType(StringType())),
StructField("stack", ArrayType(StringType())),
])
data= [(self.name, self.role, self.education, self.interests, self.stack)]
self.df_profile= (
self.spark.createDataFrame(data, schema=schema)
.withColumn("interests", F.array_join("interests", ", "))
.withColumn("stack", F.array_join("stack", " | "))
.withColumn("updated_at", F.current_date())
)
defstop(self):
self.spark.stop()
defprocess_pipeline():
pipeline=ProfilePipeline()
pipeline.build_profile()
display(pipeline.df_profile)
pipeline.stop()
process_pipeline()