Issue in extracting 54K entries into dataset and writing it into csv file in apache spark

Viewed 97

Need to write large Dataset to CSV file. Below is my sample code

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import org.apache.spark.sql.api.java.UDF3;
import org.apache.spark.sql.api.java.UDF2;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import org.apache.spark.sql.Dataset;

public class TestUdf3{
public static void main(String[] args) {
 System.setProperty("hadoop.home.dir", "F:\\JAVA\\winutils");
 JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("SparkJdbcDs").setMaster("local[*]"));
 SQLContext sqlContext = new SQLContext(sc);
 List<Row> manufactuerSynonymData = new ArrayList<Row>();

try{    
  SparkSession spark = SparkSession.builder().appName("JavaTokenizerExample").getOrCreate();
  HashMap<String, String> options = new HashMap<String, String>();options.put("header", "true");options.put("path", "D:\\xls\\Source25K.csv");  //Load source excel file
  Dataset<Row> SourcePropertSet = sqlContext.load("com.databricks.spark.csv", options) ;


 Resource resource = new ClassPathResource("/ActaulManufacturerSynonym.properties");
 Properties allProperties = PropertiesLoaderUtils.loadProperties(resource);   
 StructType schemaManufactuerSynonymDictionary = new StructType(new StructField[] {new StructField("ManufacturerSynonymSource", DataTypes.StringType, false, Metadata.empty()), new StructField("ManufacturerSynonymTarget", DataTypes.StringType, false, Metadata.empty()) });


     Set<String> setuniqueManufacturerEntries=allProperties.stringPropertyNames();
     Row individualRowEntry;
     for (String individualManufacturerEntry : setuniqueManufacturerEntries) {
          individualRowEntry=RowFactory.create(individualManufacturerEntry,allProperties.getProperty(individualManufacturerEntry));
          manufactuerSynonymData.add(individualRowEntry);
     }

     Dataset<Row> SynonaymList = spark.createDataFrame(manufactuerSynonymData, schemaManufactuerSynonymDictionary).withColumn("ManufacturerSynonymSource", lower(col("ManufacturerSynonymSource")));
     SynonaymList.show(90,false);

     UDF2<String, String, Boolean> contains = new UDF2<String, String, Boolean>() {
     private static final long serialVersionUID = -5239951370238629896L;

     @Override
     public Boolean call(String t1, String t2) throws Exception {

               return t1.matches(t2);
     }
   };

     spark.udf().register("contains", contains, DataTypes.BooleanType);

     UDF3<String, String, String, String> replaceWithTerm = new UDF3<String, String, String, String>() {

     private static final long serialVersionUID = -2882956931420910207L;

     @Override

     public String call(String t1, String t2, String t3) throws Exception {

         return t1 .replaceAll(t2,t3);         
     }
 };
     spark.udf().register("replaceWithTerm", replaceWithTerm, DataTypes.StringType);

     Dataset<Row> joined = SourcePropertSet.join(SynonaymList, callUDF("contains", SourcePropertSet.col("manufacturersource"), SynonaymList.col("ManufacturerSynonymSource"))).withColumn("ManufacturerSource", callUDF("replaceWithTerm",SourcePropertSet.col("manufacturersource"),SynonaymList.col("ManufacturerSynonymSource"), SynonaymList.col("ManufacturerSynonymTarget")));
     joined.show(54000);
     joined.repartition(1).select("*").write().format("com.databricks.spark.csv").option("delimiter", ",")
     .option("header", "true")
     .option("treatEmptyValuesAsNulls", "true")  
     .option("nullValue", "")  
     .save("D:\\xls\\synonym.csv");
  }     
  catch(Exception e){
   e.printStackTrace();
   }
  }
 }

In the above code rather than displaying the output in console using Statement :
joined.show(54000,false);

I need to write it in to csv file directly

It gives me an runtime exception they are:
1. save("D:\xls\synonym.csv");

org.apache.spark.SparkException: Job aborted.


Caused by:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 3.0 failed 1 times, most recent failure: Lost task 0.0 in stage 3.0 (TID 3, localhost, executor driver): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$apply$2: (string, string) => boolean)



2. return t1.matches(t2);

java.lang.NullPointerException


Caused by:

org.apache.spark.SparkException: Failed to execute user defined function($anonfun$apply$2: (string, string) => boolean)

Can anybody suggest how to write large dataset to excel file

0 Answers
Related