<jupyter_start><jupyter_text>How to write results into HDFSAfter submitting a job, we will need to retrieve the result. This can be stored in HDFS or elsewhere. Depending on the output size this can be a convenient approach or not. If so, we will need to write it in some format in order to read it back afterwards.To produce this test I'm using Spark 1.5.1 (Pyspark 1.5.1) and ```spark-avro``` libraries loaded like this:```bashspark-submit --packages com.databricks:spark-avro_2.10:2.0.1 [...]```For Spark 1.3.0 use```bashspark-submit --packages com.databricks:spark-avro_2.10:1.0.0 [...]```Example with Spark 1.3.0 is provided in a separated file.Index:* [How to store aggregation results](Aggregation-example) * [Example 1](First-example) * [Example 2](Second-example)* How to store selection results * Example 1<jupyter_code># is SparkContext already loaded?
sc
# Make sure you have a HiveContext
sqlContext
# Which is the version?
sc.version
# load a dataframe from Avro files
df = sqlContext.read.format("com.databricks.spark.avro").load("/cms/wmarchive/test/avro/2016/01/01/")
df.printSchema()
%%time
df.count()<jupyter_output>CPU times: user 5 ms, sys: 3 ms, total: 8 ms
Wall time: 14.8 s<jupyter_text>Aggregation examples First example1) Aggregated sum of all ```steps.performance.cpu``` values. In this case the result is a single line that can be easily stored back in HDFS, also in a textual format.<jupyter_code>aggregation1 = df.select("steps.performance.cpu") \
    .rdd \
    .flatMap(lambda cpuArrayRows: cpuArrayRows[0]) \
    .map(lambda row: row.asDict()) \
    .flatMap(lambda rowDict: [(k,v) for k,v in rowDict.iteritems()]) \
    .reduceByKey(lambda x,y: x+y)
%%time
aggregation1.collect()
# Store the file as a simple text file
aggregation1.saveAsTextFile("wmarchive/test-plaintext-aggregation1")
%%bash
hadoop fs -text wmarchive/test-plaintext-aggregation1/*
aggregated1DF = sqlContext.createDataFrame([{v[0]:v[1] for v in aggregation1.collect()}])
# saving in Json format
aggregated1DF.toJSON().saveAsTextFile("wmarchive/test-json-aggregation1")
%%bash
hadoop fs -text wmarchive/test-json-aggregation1/*
# how to write in Avro format
aggregated1DF.write.format("com.databricks.spark.avro").save("wmarchive/test-avro-aggregation1")
%%bash
hadoop fs -text wmarchive/test-avro-aggregation1/*<jupyter_output>{"AvgEventCPU":{"double":299587.125072929},"AvgEventTime":{"double":300340.72760207055},"MaxEventCPU":{"double":1351100.0},"MaxEventTime":{"double":1656177.8017510779},"MinEventCPU":{"double":1349112.0},"MinEventTime":{"double":300451.93912530516},"TotalEventCPU":{"long":151599144},"TotalJobCPU":{"double":1.520973474204235E8},"TotalJobTime":{"double":1652265.155986641}}
{'fraction_non_alphanumeric': 0.14058453570107288, 'fraction_numerical': 0.06733259341472438, 'mean_word_length': 5.828282828282828, 'pattern_counts': {'":': 18, '<': 7, '<?xml version=': 0, '>': 7, '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}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '603937', 'license': 'MIT', 'n_tokens_mistral': 1027, 'n_tokens_neox': 884, 'n_words': 239, 'path': 'How to write results into HDFS - example.ipynb', 'repo_name': 'H4ml3t/wmarchive-examples', 'size': '2703'}