I am trying to write the following spark dataset into a hive table via DataFrameWriter and the jdbc method but am getting this error:
org.apache.spark.sql.AnalysisException: Column "Field_3_date" not found in schema Some(StructType(StructField(Field_3_date,DateType,true),StructField(Field_2_integer,IntegerType,true),StructField(Field_1_string,StringType,true)));
The below is my code for how the dataset is created. Why isn't this column name recognized within the schema? Is it due to some incompatibility with how the dataset schema is defined using struct types and struct fields? If so, how to resolve this issue and is there an alternative way to write this dataset to hive without using the dataframe writer jdbc method?
SparkSession spark = SparkSession.builder().appName("documentation").master("local").getOrCreate();
String[] arr = {"abc", "1", "2021-10-10", "xyz", "2", "2020-08-23", "vw", "3", "2022-07-20", "stu", "4", "2021-09-29"};
List<String> data = Arrays.asList(arr);
List<DataType> datatype = new ArrayList<DataType>();
datatype.add(DataTypes.StringType);
datatype.add(DataTypes.StringType);
datatype.add(DataTypes.StringType);
List<String> headerList = new ArrayList<String>();
headerList.add("Field_1_string");
headerList.add("Field_2_string");
headerList.add("Field_3_string");
StructField structField1 = new StructField(headerList.get(0), datatype.get(0), true, org.apache.spark.sql.types.Metadata.empty());
StructField structField2 = new StructField(headerList.get(1), datatype.get(1), true, org.apache.spark.sql.types.Metadata.empty());
StructField structField3 = new StructField(headerList.get(2), datatype.get(2), true, org.apache.spark.sql.types.Metadata.empty());
List<StructField> structFieldsList = new ArrayList<StructField>();
structFieldsList.add(structField1);
structFieldsList.add(structField2);
structFieldsList.add(structField3);
StructType schema = new StructType(structFieldsList.toArray(new StructField[0]));
List<Row> ls = new ArrayList<Row>();
List<List<String>> l = Lists.partition(data, 3);
for(List<String> s:l) {
Row row = RowFactory.create(s.toArray());
ls.add(row);
}
Dataset<Row> dataset = spark.createDataFrame(ls, schema);
Dataset<Row> dataset2 = dataset.withColumn("Field_2_integer", dataset.col("Field_2_string").cast(DataTypes.IntegerType)).drop("Field_2_string");
Dataset<Row> dataset3 = dataset2.withColumn("Field_3_date", dataset2.col("Field_3_string").cast(DataTypes.DateType)).drop("Field_3_string");
Dataset<Row> dataset4 = dataset3.select(dataset3.col("Field_3_date"),dataset3.col("Field_2_integer"),dataset3.col("Field_1_string"));
dataset4.show();
dataset4.printSchema();
+------------+---------------+--------------+
|Field_3_date|Field_2_integer|Field_1_string|
+------------+---------------+--------------+
| 2021-10-10| 1| abc|
| 2020-08-23| 2| xyz|
| 2022-07-20| 3| vw|
| 2021-09-29| 4| stu|
+------------+---------------+--------------+
root
|-- Field_3_date: date (nullable = true)
|-- Field_2_integer: integer (nullable = true)
|-- Field_1_string: string (nullable = true)