Löschen von Zeilen in Apache Funke mit Java

Question

Müssen zweite Zeile in diesem Datensatz löschen Ich bin neu in Apache Funke kann mir jemand helfen, zu lösen. unten ist der Code:

public class DeleteRow { 
     public static void main(String[] args) { 
      System.setProperty("hadoop.home.dir", "C:\\winutils"); 
      JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("JoinFunctions").setMaster("local[*]")); 
      SQLContext sqlContext = new SQLContext(sc); 
      SparkSession spark = SparkSession.builder().appName("JavaTokenizerExample").getOrCreate(); 

      List<Row> data = Arrays.asList(
      RowFactory.create(1,"Hi I heard about Spark"), 
      RowFactory.create(2,"I wish Java could use case classes"), 
      RowFactory.create(3,"Logistic,regression,models,are,neat")); 

      StructType schema = new StructType(new StructField[] { 
      new StructField("label", DataTypes.IntegerType, false, 
       Metadata.empty()), 
      new StructField("sentence", DataTypes.StringType, false, 
      Metadata.empty()) }); 

      String ins = data.get(1).toString(); 
      System.out.println(ins); 



      Dataset<Row> sentenceDataFrame = spark.createDataFrame(data, schema); 
     sentenceDataFrame.drop(data.get(1).toString()); 

Jede Hilfe sehr geschätzt.

Answer 1

Verwenden Sie eine Filterfunktion.

Ich schrieb einen JUnit-Test, um Ihnen zu helfen.

@Test 
public void sampleDeleteRowTest() throws Exception { 
    List<Row> data = Arrays.asList(
      RowFactory.create(1, "Hi I heard about Spark"), 
      RowFactory.create(2, "I wish Java could use case classes"), 
      RowFactory.create(3, "Logistic,regression,models,are,neat")); 

    StructType schema = new StructType(new StructField[]{ 
      new StructField("label", DataTypes.IntegerType, false, 
        Metadata.empty()), 
      new StructField("sentence", DataTypes.StringType, false, 
        Metadata.empty())}); 

    String ins = data.get(1).toString(); 
    System.out.println(ins); 


    Dataset<Row> sentenceDataFrame = spark.createDataFrame(data, schema); 
    long size = sentenceDataFrame.count(); 

    assertTrue("SentenceDataFrame Size = " + sentenceDataFrame.count(), size == 3); 

    sentenceDataFrame = sentenceDataFrame.filter(new FilterFunction<Row>() { 
     @Override 
     public boolean call(Row row) throws Exception { 
      Integer label = row.getInt(0); 
      return label != 2; 
     } 
    }); 

    size = sentenceDataFrame.count(); 

    assertTrue("SentenceDataFrame Size = " + sentenceDataFrame.count(), size == 2); 
} 

https://spark.apache.org/docs/2.1.0/api/java/org/apache/spark/sql/Dataset.html#filter(org.apache.spark.api.java.function.FilterFunction)

https://spark.apache.org/docs/2.1.0/api/java/org/apache/spark/api/java/function/FilterFunction.html

Alternativ können Sie auch eine Lambda-Funktion verwenden, um die gleiche gewünschte Ergebnis zu erzielen:

sentenceDataFrame = sentenceDataFrame.filter((FilterFunction<Row>) r -> (r).getInt(0) != 2);