fqaiser94 / mse Goto Github PK

import com.github.fqaiser94.mse.methods._
import za.co.absa.spark.hofs._
import spark.implicits._

val jsonText = """{
     |   "data1": [
     |     {
     |       "vlan": {
     |         "id": "195",
     |         "name": "Subnet-54.14.195"
     |       }
     |     },
     |     {
     |       "vlan": {
     |         "id": "195",
     |         "name": "Subnet-54.14.193"
     |       }
     |     }
     |   ]
     | }""".stripMargin

val df = spark.read.option("multiline", "true").json(Seq(jsonText).toDS())

# this works fine
df.withColumn("data1", transform($"data1", col => col.withFieldRenamed("vlan", "vlan_renamed"))).printSchema

# but what if I want to rename "name" to "device_name" ?
# I thought it would be something like this
df.withColumn("data1", transform($"data1", col => col.withField("vlan", $"data1.vlan".withFieldRenamed("name", "device_name")))).printSchema
# but it fails with:
org.apache.spark.sql.AnalysisException: cannot resolve 'rename_fields(`data1`.`vlan`, 'name', 'device_name')' due to data type mismatch: Only struct is allowed to appear at first position, got: array.;;
'Project [transform(data1#44, lambdafunction(add_fields(lambda elm#58, vlan, rename_fields(data1#44.vlan, name, device_name)), lambda elm#58, false)) AS data1#57]
+- LogicalRDD [data1#44], false

from datetime import datetime

from mse import *

from pyspark.sql import SparkSession
from pyspark.sql import functions as f
from pyspark.sql.types import StructType, StringType, TimestampType, IntegerType


def spark():
    """
    This function is invoked to create the Spark Session.

    :return: the spark session
    """
    spark_session = (SparkSession
                     .builder
                     .appName("Data_Experimentation_Framework")
                     .getOrCreate())

    spark_session.conf.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MILLIS")
    return spark_session


address = StructType().add("state", StringType())
schema = StructType().add("_id", IntegerType()) \
    .add("employer", StringType()) \
    .add("created_at", TimestampType()) \
    .add("name", StringType())\
    .add("address", address)

employees = [{'_id': 1,
              'employer': 'Microsoft',
              'created_at': datetime.now(),
              'name': 'Noel',
              'address': {
                  "state": "Pennsylvania"
              }

              },
             {'_id': 2,
              'employer': 'Apple',
              'created_at': datetime.now(),
              'name': 'Steve',
              'address': {
                  "state": "New York"
              }
              }
             ]
df = spark().createDataFrame(employees, schema=schema)

df.withColumn("address", f.col("address").withField("country", f.lit("USA")))

df.printSchema()

df.write \
    .format("parquet") \
    .mode("append") \
    .save("/Users/felixkizhakkeljose/Downloads/test12")