Wie man herausgefilterte jsons in der Spark-Filterfunktion

Question

sammelt oder speichert Ich möchte die herausgefilterten Daten speichern oder sammeln, d. H. Die jsons, die die Validierungen für hdfs oder hbase nicht bestehen.

dstream.filter { data => VitalValidator.isVitalJSONValid(data) } 

wo DSTREAM ist DSTREAM [Zeichenfolge] und isVitalJSONValid akzeptiert String und gibt boolean

Answer 1

ich so etwas wie dies mit Scala tun würde.

def isVitalJSONValid(data: String): Boolean = { 
    var isValid = false 

    //peroforms some validation 
    if(data.equals("some/validation")){ 
    isValid = true 
    } 
    !isValid 
} 


//existing goes on here 
dstream.filter(data => isVitalJSONValid(data)).saveAsHadoopFiles("file_prefix") 

Output operations on dstreams