<commit_before>from optparse import OptionParser
from pyspark import SparkContext, SparkConf, StorageLevel
import json
from etk.core import Core
import codecs
__author__ = 'amandeep'


needed_fields = ['raw_content', 'content_extraction', 'knowledge_graph', 'url', 'timestamp', 'tld']


def remove_junk(x):
    # new_x = dict()
    # if 'doc_id' in x:
    #     new_x['doc_id'] = x['doc_id']
    if '_id' in x:
        x['doc_id'] = x['_id']

    # for field in needed_fields:
    #     if field in x:
    #         new_x[field] = x[field]
    return x

if __name__ == '__main__':
    compression = "org.apache.hadoop.io.compress.GzipCodec"

    parser = OptionParser()
    (c_options, args) = parser.parse_args()
    input_path = args[0]
    output_path = args[1]
    extraction_config_path = args[2]

    sc = SparkContext(appName="ETK-Extractions")
    conf = SparkConf()
    extraction_config = json.load(codecs.open(extraction_config_path))
    c = Core(extraction_config=extraction_config)
    input_rdd = sc.sequenceFile(input_path).partitionBy(1000)
    input_rdd = input_rdd.mapValues(json.loads).mapValues(
        lambda x: c.process(x, create_knowledge_graph=True)).mapValues(remove_junk).mapValues(json.dumps)
    input_rdd.saveAsSequenceFile(output_path, compressionCodecClass=compression)
<commit_msg>add doc id and filter pages with no html<commit_after>from optparse import OptionParser
from pyspark import SparkContext, SparkConf, StorageLevel
import json
from etk.core import Core
import codecs
__author__ = 'amandeep'


needed_fields = ['raw_content', 'content_extraction', 'knowledge_graph', 'url', 'timestamp', 'tld']


def add_doc_id(x):
    if '_id' in x:
        x['doc_id'] = x['_id']
    return x


def remove_if_no_html(x):
    if 'raw_content' not in x:
        return False
    rc = x['raw_content']
    if not rc:
        return False
    if rc.strip() == '':
        return False
    return True

if __name__ == '__main__':
    compression = "org.apache.hadoop.io.compress.GzipCodec"

    parser = OptionParser()
    (c_options, args) = parser.parse_args()
    input_path = args[0]
    output_path = args[1]
    extraction_config_path = args[2]

    sc = SparkContext(appName="ETK-Extractions")
    conf = SparkConf()
    extraction_config = json.load(codecs.open(extraction_config_path))
    c = Core(extraction_config=extraction_config)
    input_rdd = sc.sequenceFile(input_path).partitionBy(1000)
    input_rdd = input_rdd.mapValues(json.loads).filter(lambda x: remove_if_no_html(x[1])).mapValues(add_doc_id)\
        .mapValues(lambda x: c.process(x, create_knowledge_graph=True)).mapValues(json.dumps)
    input_rdd.saveAsSequenceFile(output_path, compressionCodecClass=compression)

{'fraction_non_alphanumeric': 0.11650128629180448, 'fraction_numerical': 0.005512679162072767, 'mean_word_length': 4.106941838649155, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1599493', 'n_tokens_mistral': 947, 'n_tokens_neox': 924, 'n_words': 192}