J'ai un grand fichier de données XML (> 160m) à traiter, et il semble que la solution SAX / Expat / Pattdom est la voie à suivre. Je voudrais avoir un fil qui tamise à travers les nœuds et pousse les nœuds à traiter sur une file d'attente, puis d'autres threads de travailleur tirent le prochain nœud disponible hors de la file d'attente et le traitent.
J'ai ce qui suit (il devrait avoir des serrures, je sais - ce sera, plus tard) p> Le problème est que le corps du bloc Je suis sûr que je suis sûr C'est ma propre ignorance, mais je ne vois pas où je fais l'erreur. p> ps: j'ai aussi essayé de changer pendant code> est appelé une seule fois, puis Je ne peux même pas l'interrompre Ctrl-C. Sur des fichiers plus petits, la sortie est comme prévu, mais cela semble indiquer que le gestionnaire ne s'appelle que lorsque le document est totalement analysé, ce qui semble vaincre le but d'un analyseur SAX. P> start_handler code> Ainsi: p> def start_handler(name, attrs):
def app():
q.append(name)
u = threading.Thread(group=None, target=app)
u.start()
4 Réponses :
La seule chose que je vois est fausse, c'est que vous accédez à Essayez de verrouiller, ce n'est vraiment pas très difficile: p> q code> simultanément de différents threads - pas de verrouillage comme vous écrivez en effet. Cela demande des problèmes - et vous avez probablement des problèmes sous la forme de l'interprète Python enfermant sur vous. :) import sys, time
import xml.parsers.expat
import threading
q = []
q_lock = threading.Lock() <---
def start_handler(name, attrs):
q_lock.acquire() <---
q.append(name)
q_lock.release() <---
def do_expat():
p = xml.parsers.expat.ParserCreate()
p.StartElementHandler = start_handler
p.buffer_text = True
print("opening {0}".format(sys.argv[1]))
with open(sys.argv[1]) as f:
print("file is open")
p.ParseFile(f)
print("parsing complete")
t = threading.Thread(group=None, target=do_expat)
t.start()
while True:
q_lock.acquire() <---
print(q)
q_lock.release() <---
time.sleep(1)
Je ne suis pas trop sûr de ce problème. Je suppose que l'appel à Paysfile bloque et que seul le fil d'analyse est en cours de courir à cause de la gil. Un moyen autour de ce serait d'utiliser multiprocession à la place. Il est conçu pour travailler avec des files d'attente, de toute façon.
Vous faites un J'ai inclus une liste d'éléments, juste pour reproduire votre script d'origine. Votre solution finale utilisera probablement Processus et vous pouvez transmettre un Queue : get_nowait et un piscine ou quelque chose de similaire.
Oui, c'est un bon chemin pour descendre - comme vous l'avez dit que vous voudriez utiliser des files d'attente de toute façon.
J'ai essayé ce code; Cela évite le verrouillage, mais ParseFile ne semble toujours rien produire tant que cela n'a pas lu l'entrée entière.
Je ne connais pas beaucoup de choses sur la mise en œuvre, mais si l'analyse est Code C qui exécute jusqu'à la fin, d'autres threads Python ne seront pas exécutés. Si l'analyseur rappelle au code Python, le GIL peut être libéré pour d'autres threads à courir, mais je ne suis pas sûr de cela. Vous voudrez peut-être vérifier ces détails.
the Le deuxième point est, oubliez les serrures pour cette utilisation! - Utilisez Queue.Queue à la place, il est intrinsèquement threadsafe et presque invariablement le moyen le plus simple et le plus simple de coordonner plusieurs threads en python. Il suffit de faire une file d'attente instance q , (il existe plusieurs stratégies auxiliaires que vous pouvez utiliser pour coordonner la résiliation des threads de travailleurs quand il n'y a plus de travail pour eux Faites, mais les exigences particulières les plus simples et absentes, consistent simplement à les faire des fils de démon, de sorte qu'ils terminent tous lorsque le fil principal fait - voir Les docs ). Paysfile , comme vous l'avez remarqué, il suffit de "gorgers" "- pas de bien pour le incrémental analysant que vous voulez faire! Donc, il suffit de nourrir le fichier à l'analyseur un peu à la fois, en veillant à donner un contrôle de manière conditionnelle à d'autres threads, par exemple: temps.sleep ( 0.0) L'appel est le moyen de Python de dire "Rendement à d'autres threads s'il est prêt et en attente"; La méthode parse est documentée ici . q.put (nom) dessus et avez-vous travaillé le bloc de threads sur q.get ( ) en attente de faire plus de travail à faire - c'est tellement simple!
Votez-vous pour les suggestions de la file d'attente, mais êtes-vous sûr de savoir à propos de Paysfile, tout en tombent en une fois? Il rappelle dans les manutentionnaires de Python pour gérer les balises comme ça se passe, c'est tout le but de la solution saxe ... Ou dites-vous que cela ne suffit pas pour déclencher un interrupteur à fil en python?
Si vous voulez SAX, vous pouvez utiliser XML.Sax, voir docs.python.org/library/... ; L'OP n'utilise pas saz, mais plutôt xml.parsers.expat, une interface d'abstraction inférieure qui ne pas impose une stratégie incrémentielle (elle prend en charge elle, mais pas < I> Impose , laissez-le au niveau du code Python de choisir et de choisir).
Le choix de l'expatrié était quelque peu arbitraire, je n'ai pas pu trouver une bonne explication de la différence entre Expat & SAX. Le module SAX fonctionne aussi bien - même peut-être mieux, car il semble être aussi asynchrone que nécessaire. J'ai chuté de toute façon l'adoption de la méthode "nourrir un morceau à la fois", car cela me donne une chance de stériliser les cordes que je nourris avant que l'analyseur ne soit envisagé. Réponse très utile, merci.
@slide_rule, vous êtes le bienvenu - et oui, SAX applique une utilisation asynchrone (entraînée par événement) et vous permet d'utiliser différents analyseurs sous-jacents (expatriés celui qui est livré avec Python, mais vous pouvez installer d'autres, par exemple ceux qui valident le WRT un schéma XML & c).