spark sql

# -*- coding: UTF-8 -*-from pyspark import SparkConf, SparkContext, StorageLevel
from pyspark.sql import Row, SQLContext
from py4j.protocol import Py4JJavaError
from collections import OrderedDict
import jsonconf = SparkConf().setAppName('SimpleApp').setMaster('local[*]')
sc = SparkContext(conf=conf)
sql_context = SQLContext(sc)  # Not to mention you need a SQLContext to work with DataFrames anyway.def try_text_file(path):# handle existing-empty-file and non-existent-filerdd = sc.textFile(path)try:rdd.first()return rddexcept Py4JJavaError:return sc.emptyRDD()s3_urls = ['172.31.19.1/QAClick-2887717633-20011-2017012800.tar.gz','172.31.0.142/QAClick-2887712910-20010-2016101203.tar.gz']
paths = ['s3a://aff.parallel-app.com/{url}'.format(url=url) for url in s3_urls]
rdd_file = sc.union([try_text_file(path) for path in paths])# s_urls = ','.join(['s3a://aff.parallel-app.com/{url}'.format(url=url) for url in tmp_urls])
# rdd_file = sc.textFile(s_urls, 2)def convert_line_to_row(line):line = line.strip()idx = line.find(' {')if idx != -1:try:d = json.loads(line[idx:])except:d = {}else:d = {}if d:ymdhms = line[:14]else:ymdhms = ''ymd = ymdhms[:8]hms = ymdhms[8:14]pkg_name = d.get('ad_pkg_name', '')perform_install = d.get('perform_install', False)if perform_install:perform_install = 1else:perform_install = 0tmp_dict = {'pkg_name': pkg_name, 'perform_install': perform_install, 'ymd': ymd, 'hms': hms}return Row(**OrderedDict(sorted(tmp_dict.items())))df = rdd_file.map(convert_line_to_row).toDF()
# df.persist(StorageLevel.MEMORY_AND_DISK)  # MEMORY_ONLY，MEMORY_AND_DISK，DISK_ONLY
# print rdd_file.take(2)df.registerTempTable("tbl_qac")
sql = """
select ymd,pkg_name,count(*)
from tbl_qac where pkg_name!='' and perform_install=1
group by ymd,pkg_name
"""
sc_sql = sql_context.sql(sql)
sc_sql.show()
print sc_sql.collect()