本文的代码涉及几个知识点，都是比较有用： 1、Spark用JAVA编写代码的方式； 2、Spark读取MySQL数据表，并且使用的是自定义SQL的方式，默认会读取整个表的； 3、Spark使用sql.functions的原有方法，给dataframe新增列、变更列； 4、Spark使用udf的自定义函数，给dataframe新增列、变更列； [crayon-5d58c0c9d5aa62988260 … 继续阅读Spark使用JAVA编写自定义函数修改DataFrame
对于PyCharm，需要作如下设置： 1、安装pyspark，它会自动安装py4j 2、在edit configuration中，add content root，选择spark下载包的python/pyspark/lib下的pyspark.zip和py4j.zip两个包； 代码实例：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 | from pyspark.sql import Row from pyspark.sql import SparkSession logFile = "file:///Users/peishuaishuai/tmp/sparktest.txt" # Should be some file on your system spark = SparkSession.builder.appName("SimpleApp").getOrCreate() input = spark.read.text(logFile).rdd.map( lambda x: str(x[0]).split("\t") ).filter( lambda x: len(x) == 2 ).map( lambda x: Row(name=x[0], grade=int(x[1])) ) schemaData = spark.createDataFrame(input) schemaData.createOrReplaceTempView("tb") print(schemaData.count()) schemaData.printSchema() datas = spark.sql("select name,sum(grade) from tb group by name").rdd.map( lambda x: "\t".join([x[0], str(x[1])]) ) datas.repartition(3).saveAsTextFile("file:///Users/peishuaishuai/tmp/sparktest_output") spark.stop() |
输入数据为： [crayon-5 … 继续阅读PyCharm开发PySpark程序的配置和实例
1、避免shuffle，改reduce join为map join，适用于JOIN的时候有一个表是小表的情况，直接使用collect()获取小表的所有数据，然后brodcast，对大表进行MAP，MAP时直接提取broadcast的小表数据实现JOIN； 2、随机数的方案，对于聚合类操作，可以分步骤进行聚合，第一步，在原来的KEY后面加上随机数（比如1~10），然后进行聚合（比如SUM操作）；第二 … 继续阅读Spark数据倾斜解决方法