How do I create a new DataFame based on an old DataFrame?

Question

I have csv file: dbname1.table1.csv:

|target            | source        |source_table                       |relation_type|
 ---------------------------------------------------------------------------------------
avg_ensure_sum_12m | inn_num       | custom_cib_ml_stg.p_overall_part_tend_cust | direct
avg_ensure_sum_12m | protocol_dttm | custom_cib_ml_stg.p_overall_part_tend_cust | direct
avg_ensure_sum_12m | inn_num       | custom_cib_ml_stg.p_overall_part_tend_cust | indirect

csv format for this table:

target,source,source_table,relation_type
avg_ensure_sum_12m,inn_num,custom_cib_ml_stg.p_overall_part_tend_cust,direct
avg_ensure_sum_12m,protocol_dttm,custom_cib_ml_stg.p_overall_part_tend_cust,direct
avg_ensure_sum_12m,inn_num,custom_cib_ml_stg.p_overall_part_tend_cust,indirect

Then I create a dataframe by reading it:

 val dfDL = spark.read.option("delimiter", ",")
                     .option("header", true)
                     .csv(file.getPath.toUri.getPath)

Now I need to create a new dataframe based on dfDL.

The structure of the new dataframe looks like this:

case class DataLink(schema_from: String,
                    table_from: String,
                    column_from: String,
                    link_type: String,
                    schema_to: String,
                    table_to: String,
                    column_to: String)

The information for the fields of the new DataFrame is obtained from a csv file:

pseudocode:
schema_from = source_table.split(".")(0) // Example: custom_cib_ml_stg
table_from  = source_table.split(".")(1) // Example: p_overall_part_tend_cust
column_from = source                     // Example: inn_num
link_type   = relation_type              // Example: direct
schema_to   = "dbname1.table1.csv".split(".")(0) // Example: dbname1
table_to    = "dbname1.table1.csv".split(".")(1) // Example: table1
column_to   = target                             // Example: avg_ensure_sum_12m

I need to create a new dataframe. I can't cope on my own.

P.S. I need this dataframe to create a json file from it later. Example JSON:

[{"schema_from":"custom_cib_ml36_stg",
"table_from":"p_overall_part_tend_cust",
"column_from":"inn_num",
"link_type":"direct",
"schema_to":"dbname1",
"table_to":"table1",
"column_to":"avg_ensure_sum_12m"
},
{"schema_from":"custom_cib_ml36_stg",
"table_from":"p_overall_part_tend_cust",
"column_from":"protocol_dttm",
"link_type":"direct","schema_to":"dbname1",
"table_to":"table1",
"column_to":"avg_ensure_sum_12m"}

I don't like my current implementation:

def readDLFromHDFS(file: LocatedFileStatus): Array[DataLink] = {

    val arrTableName        = file.getPath.getName.split("\\.")
    val (schemaTo, tableTo) = (arrTableName(0), arrTableName(1))

    val dfDL = spark.read.option("delimiter", ",")
                         .option("header", true)
                         .csv(file.getPath.toUri.getPath)

    //val sourceTable = dfDL.select("source_table").collect().map(value => value.toString().split("."))

    dfDL.collect.map(row => DataLink(row.getString(2).split("\\.")(0),
                                     row.getString(2).split("\\.")(1),
                                     row.getString(1),
                                     row.getString(3),
                                     schemaTo,
                                     tableTo,
                                     row.getString(0)))
  }

  def toJSON(dataLinks: Array[DataLink]): Option[JValue] =
    dataLinks.map(Extraction.decompose).reduceOption(_ ++ _)

}

rudolfovic · Accepted Answer · 2021-12-05T21:02:44.643

1

You definitely don't want to collect, that defeats the point of using spark here. As always with Spark you have a lot of options. You can use RDDs but I don't see a need to switch between modes here. You just want to apply custom logic to some columns and end up with a dataframe with the resulting column alone.

First, define a UDF that you want to apply:

def convert(target, source, source_table, relation_type) =
  DataLink(source_table.split("\\.")(0),
           source_table.split("\\.")(1),
           source,
           "dbname1.table1.csv".split(".")(0)
           "dbname1.table1.csv".split(".")(1)
           target))

Then apply this function to all the relevant columns (making sure you wrap it in udf to make it a spark function rather than a plain Scala function) and select the result:

df.select(udf(convert)($"target", $"source", $"source_table", $"relation_type"))

If you want to a DataFrame with 7 columns as your result:

df.select(
  split(col("source_table"), "\\.").getItem(0),
  split(col("source_table"), "\\.").getItem(1),
  col("source"),
  lit("dbname1"),
  lit("table1"),
  col("target")
)

You can also add .as("column_name") to each of these 7 columns.

edited Dec 05 '21 at 21:02

answered Dec 02 '21 at 23:48

rudolfovic

3,163
2
14
38

Hi! This is cool solution. Very easy for reading. But i have small problem. What am I doing wrong? https://pic4a.ru/112/JYd.png – Mardaunt Dec 03 '21 at 12:23
What exactly is the problem? – rudolfovic Dec 03 '21 at 12:37
sorry =) "Value '$' is not a member of StringContext" This problem https://pic4a.ru/112/kL2.png – Mardaunt Dec 03 '21 at 13:18
1

not sure why you get this but you can also use `col("target")` instead – rudolfovic Dec 03 '21 at 13:24
Ohh. col("target") is working. But one more problem remains. https://pic4a.ru/112/u6E.png – Mardaunt Dec 03 '21 at 13:42
Cannot resolve overloaded method 'select' – Mardaunt Dec 03 '21 at 13:44
1

Try to store `udf(convert)` in a separate variable maybe? and then apply `f(col("...` – rudolfovic Dec 03 '21 at 14:19
Thanks! New variable for udf function work for me. All good. But i get DataFrame with the one column. How i can create DataFrame with 7 column? I publish answer with my code. – Mardaunt Dec 05 '21 at 20:56
Then why do you need the `DataLink` class? You can instead have 7 separate columns in your select (instead of one which is the result of the udf application) – rudolfovic Dec 05 '21 at 20:58
Hmm... Ok! I try – Mardaunt Dec 05 '21 at 21:12
I wanted to use DataLink to assemble an array of DataLink objects and then create JSON objects from them. Maybe I need to create a JSON from a DataFrame – Mardaunt Dec 05 '21 at 21:27
I don't see a need for this object - check this out https://stackoverflow.com/questions/36157810/spark-row-to-json also consider accepting the answer :) – rudolfovic Dec 05 '21 at 21:32
I received a very detailed consultation. Your answer is very comprehensive. !! – Mardaunt Dec 05 '21 at 21:57

score 1 · Answer 2 · answered Dec 02 '21 at 23:59

You can use dataset directly.

import spark.implicits._

case class DataLink(schema_from: String,
                    table_from: String,
                    column_from: String,
                    link_type: String,
                    schema_to: String,
                    table_to: String,
                    column_to: String)

val filename = "dbname1.table1.csv"
val df = spark.read.option("header","true").csv("test.csv")
df.show(false)
+------------------+-------------+------------------------------------------+-------------+
|target            |source       |source_table                              |relation_type|
+------------------+-------------+------------------------------------------+-------------+
|avg_ensure_sum_12m|inn_num      |custom_cib_ml_stg.p_overall_part_tend_cust|direct       |
|avg_ensure_sum_12m|protocol_dttm|custom_cib_ml_stg.p_overall_part_tend_cust|direct       |
|avg_ensure_sum_12m|inn_num      |custom_cib_ml_stg.p_overall_part_tend_cust|indirect     |
+------------------+-------------+------------------------------------------+-------------+

df.createOrReplaceTempView("table")

val df2 = spark.sql(s"""
select split(source_table, '[.]')[0] as schema_from
     , split(source_table, '[.]')[1] as table_from
     , source                        as column_from
     , relation_type                 as link_type
     , split('${filename}', '[.]')[0] as schema_to
     , split('${filename}', '[.]')[1] as table_to
     , target                        as column_to
  from table
""").as[DataLink]

df2.show()

+-----------------+--------------------+-------------+---------+---------+--------+------------------+
|      schema_from|          table_from|  column_from|link_type|schema_to|table_to|         column_to|
+-----------------+--------------------+-------------+---------+---------+--------+------------------+
|custom_cib_ml_stg|p_overall_part_te...|      inn_num|   direct|  dbname1|  table1|avg_ensure_sum_12m|
|custom_cib_ml_stg|p_overall_part_te...|protocol_dttm|   direct|  dbname1|  table1|avg_ensure_sum_12m|
|custom_cib_ml_stg|p_overall_part_te...|      inn_num| indirect|  dbname1|  table1|avg_ensure_sum_12m|
+-----------------+--------------------+-------------+---------+---------+--------+------------------+

score 0 · Answer 3 · answered Dec 05 '21 at 21:05

My progress... Now, i can create new DataFrame, but he contain only 1 column.

val dfDL = spark.read.option("delimiter", ",")
                     .option("header", true)
                     .csv(file.getPath.toUri.getPath)

val convertCase = (target: String, source: String, source_table: String, relation_type: String) =>
                    DataLink(
                      source_table.split("\\.")(0),
                      source_table.split("\\.")(1),
                      source,
                      relation_type,
                      schemaTo,
                      tableTo,
                      target,
                    )


val udfConvert = udf(convertCase)

val dfForJson  = dfDL.select(udfConvert(col("target"),
                                        col("source"),
                                        col("source_table"),
                                        col("relation_type")))

How do I create a new DataFame based on an old DataFrame?

3 Answers3