PySpark Zählen Zeilen, die Zeichenfolge enthalten

Question

ich mehrere XML-Dateien, die wie folgt aussehen:

<?xml version="1.0" encoding="UTF-8"?> 
<parent> 
    <row AcceptedAnswerId="15" AnswerCount="5" Body="&lt;p&gt;How should 
I elicit prior distributions from experts when fitting a Bayesian 
model?&lt;/p&gt;&#10;" CommentCount="1" CreationDate="2010-07- 
19T19:12:12.510" FavoriteCount="17" Id="1" LastActivityDate="2010-09- 
15T21:08:26.077" OwnerUserId="8" PostTypeId="1" Score="26" 
Tags="&lt;bayesian&gt;&lt;prior&gt;&lt;elicitation&gt;" 
Title="Eliciting priors from experts" ViewCount="1457" /> 

Ich mag wäre in der Lage sein PySpark zu verwenden, um die Zeilen zu zählen, die die Zeichenfolge nicht enthalten: <row

Mein aktueller Gedanke:

def startWithRow(line): 
    if line.strip().startswith("<row"): 
     return True 
    else: 
     return False 

sc.textFile(localpath("folder_containing_xmg.gz_files")) \ 
    .filter(lambda x: not startWithRow(x)) \ 
    .count() 

ich diese Validierung haben versucht, aber ich Ergebnisse von selbst eine einfache Zählung Linien bekommen, dass ich keinen Sinn machen (heruntergeladen die xml Datei und machte eine wc darauf, die nicht die Wortzahl von PySpark zusammengehörte.)

Fällt irgendetwas über meine Annäherung oben als falsch/sonderbar heraus?

Answer 1

Ich werde einfach lxml Bibliothek kombiniert mit Spark verwenden, um die Zeile mit row zu zählen oder etwas herauszufiltern.

from lxml import etree 

def find_number_of_rows(path): 
    try: 
     tree = etree.fromstring(path) 
    except: 
     tree = etree.parse(path) 
    return len(tree.findall('row')) 

rdd = spark.sparkContext.parallelize(paths) # paths is a list to all your paths 
rdd.map(lambda x: find_number_of_rows(x)).collect() 

Zum Beispiel, wenn Sie Liste oder XML-String haben (nur Spielzeug Beispiel), können Sie wie folgt vorgehen:

text = [ 
    """ 
    <parent> 
     <row ViewCount="1457" /> 
     <row ViewCount="1457" /> 
    </parent> 
    """, 
    """ 
    <parent> 
     <row ViewCount="1457" /> 
     <row ViewCount="1457" /> 
     <row ViewCount="1457" /> 
    </parent> 
    """ 
] 

rdd = spark.sparkContext.parallelize(text) 
rdd.map(lambda x: find_number_of_rows(x)).collect() 

In Ihrem Fall Ihre Funktion in Weg zu nehmen, anstatt eine Datei . Dann können Sie diese Zeilen zählen oder filtern. Ich habe keine vollständige Datei zum Testen. Lassen Sie mich wissen, wenn Sie zusätzliche Hilfe benötigen!

Answer 2

def badRowParser(x):  
    try: 
     line = ET.fromstring(x.strip().encode('utf-8')) 
     return True 
    except: 
     return False 
posts = sc.textFile(localpath('folder_containing_xml.gz_files')) 
rejected = posts.filter(lambda l: "<row" in l.encode('utf- 
8')).map(lambda x: not badRowParser(x)) 
ans = rejected.collect() 

from collections import Counter 
Counter(ans)