spark的一些学习记录（持续）

from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .master("local") \
    .appName("Word Count") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

data=sc.paralleize(['1',2],{'2':3},['string'])

def vals_concat(v1, v2):
  return v1 + ',' + v2
input5 = input4.reduceByKey(vals_concat)

import os 
import sys
spark_name = os.




    
environ.get('SPARK_HOME',None)
# SPARK_HOME即spark的安装目录，不用到bin级别，一般为/usr/local/spark
if not spark_home:
    raise ValueErrorError('spark 环境没有配置好')
# sys.path是Python的第三方包查找的路径列表，将需要导入的包的路径添加进入，避免 can't find modal xxxx
# 这个方法应该同 spark-submit提交时添加参数 --py_files='/path/to/my/python/packages.zip',将依赖包打包成zip 添加进去 效果一致
sys.path.insert(0,'/root/virtualenvs/my_envs/lib/python3.6/site-packages/')
sys.path.insert(0,os.path.join(spark_name,'python')
sys.path.insert(0,os.path.join(spark_name,'python/lib/py4j-0.10.7-src.zip'))
# sys.path.insert(0,os.path.join(spark_name,'libexec/python'))
# sys.path.insert(0,os.path.join(spark_name,'libexex/python/build'))
from pyspark import SparkConf, SparkContext
#这样就可以成功导入 SparkConf,SparkContext了，而不会出现找不到SparkConf和SparkContext的奇怪问题了。
import os
os.environ["PYSPARK_PYTHON"] = "./python36/pyspark_py3/bin/python"
os.environ["PYSPARK_DRIVER_PYTHON"] = "some conda python env"

from pyspark.sql import SparkSession
spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

# spark is an existing SparkSession
df = spark.read.json("examples/src/main/resources/people.json")
# Displays the content of the DataFrame to stdout
df.show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

# spark, df are from the previous example
# Print the schema in a tree format
df.printSchema()
# root
# |-- age: long (nullable = true)
# |-- name: string (nullable = true)
# Select only the "name" column
df.select("name").show()
# +-------+
# |   name|
# +-------+
# |Michael|
# |   Andy|
# | Justin|
# +-------+
# Select everybody, but increment the age by 1
df.select(df['name'], df['age'] + 1).show()
# +-------+---------+
# |   name|(age + 1)|
# +-------+---------+
# |Michael|     null|
# |   Andy|       31|
# | Justin|       20|
# +-------+---------+
# Select people older than 21
df.filter(df['age'] > 21).show()
# +---+----+
# |age|name|
# +---+----+
# | 30|Andy|
# +---+----+
# Count people by age
df.groupBy("age").count().show()
# +----+-----+
# | age|count|
# +----+-----+
# |  19|    1|
# |null|    1|
# |  30|    1|
# +----+-----+

# Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people")
sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

df.createGlobalTempView("people")
# Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+
# Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

from pyspark.sql import Row
sc = spark.sparkContext
# Load a text file and convert each line to a Row.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))
# Infer the schema, and register the DataFrame as a table.
schemaPeople = spark.createDataFrame(people)
schemaPeople.createOrReplaceTempView("people")
# SQL can be run over DataFrames that have been registered as a table.
teenagers = spark.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
# The results of SQL queries are Dataframe objects.
# rdd returns the content as an :class:`pyspark.RDD` of :class:`Row`.
teenNames = teenagers.rdd.map(lambda p: "Name: " + p.name).collect()
for name in teenNames:
    print(name)
# Name: Justin

# Import data types
from pyspark.sql.types import StringType, StructType, StructField
sc = spark.sparkContext
# Load a text file and convert each line to a Row.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
# Each line is converted to a tuple.
people = parts.map(lambda p: (p[0], p[1].strip()))
# The schema is encoded in a string.
schemaString = "name age"
fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
schema = StructType(fields)
# Apply the schema to the RDD.
schemaPeople = spark.createDataFrame(people, schema)
# Creates a temporary view using the DataFrame
schemaPeople.createOrReplaceTempView("people")
# SQL can be run over DataFrames that have been registered as a table.
results = spark.sql("SELECT name FROM people")
results.show()
# +-------+
# |   name|
# +-------+
# |Michael|
# |   Andy|
# | Justin|
# +-------+

sparkDF=spark.createDataFrame(pandasDF) 
sparkDF.printSchema()
sparkDF.show()

from pyspark.sql.types import StructType,StructField, StringType, IntegerType
#Create User defined Custom Schema using StructType
mySchema = StructType([ StructField("First Name", StringType(), True)\
                       ,StructField("Age", IntegerType(), True)])
#Create DataFrame by changing schema
sparkDF2 = spark.createDataFrame(pandasDF,schema=mySchema)
sparkDF2.printSchema()
sparkDF2.show()

from pyspark.sql import SparkSession, HiveContext
_SPARK_HOST = "spark://spark-master:7077"
_APP_NAME = "test"
spark = SparkSession.builder.master(_SPARK_HOST).appName(_APP_NAME).getOrCreate()
data = [
 (1,"3","145"),
 (1,"4","146"),
 (1,"5","25"),
 (1,"6","26"),
 (2,"32","32"),
 (2,"8","134"),
 (2,"8","134"),
 (2,"9","137")
df = spark.createDataFrame(data, ['id', "test_id", 'camera_id'])
# method one，default是默认数据库的名字，write_test 是要写到default中数据表的名字
df.registerTempTable('test_hive')
sqlContext.sql("create table default.write_test select * from test_hive")

df.write.format("hive").mode("overwrite").saveAsTable('default.write_test')

df_write.write.format("parquet").mode("overwrite")
.saveAsTable("course_table") ## 第一次构建的时候这么写会自动创建新表，后面insertinto

df.write.format('parquet').mode('overwrite').\ 
partitionBy("channel","event_day","event_hour")\  # 分区字段逗号分割，跟建表顺序一致   
.saveAsTable("table_name")  # 如果首次执行不存在这个表，会自动创建分区表，不指定分区即创建不带分区的表

# 可动态分区设置
spark.sql("set hive.exec.dynamic.partition.mode=nonstrict;")
# df 按照字段+分区的顺序对应hive顺序
df.write\ 
.insertInto("table_name")  # 如果执行不存在这个表，会报错
pyspark之DataFrame写hive表方式_SummerHmh的博客-CSDN博客# 可动态分区设置
spark.sql("set hive.exec.dynamic.partition.mode=nonstrict;")
# df 按照字段+分区的顺序对应hive顺序
df.write\ 
.insertInto("table_name")  # 如果执行不存在这个表，会报错

data.repartition(1).write.csv(path='file:///home/wangbingnan/wangbingnan/纯注册场景/register_data',mode='ignore',header=True)

    spark = SparkSession.builder \
        .appName("graph_fraud_detections") \
        .config('spark.driver.memory', '2g') # driver最终执行的内存大小，例如to csv
这类操作，最终是collect到driver上然后to csv的，所以设定这里就是设定最终driver的计算结果的大小 \
        .config("spark.executor.memory", "2g") \
        .config("spark.driver.maxResultSize","50g") \
        .config("spark.default.parallelism", "20") \
        .enableHiveSupport() \
        .getOrCreate() 

from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .appName("address_fraud_detection") \
    .master("yarn") \
    .config("spark.submit.deployMode", "client") \
    .config("spark.yarn.queue", "XXX") \

os.environ["PYSPARK_PYTHON"] = "./python36/xxxxxx"
os.environ["ARROW_PRE_0_15_IPC_FORMAT"]="1"
os.environ['PYARROW_IGNORE_TIMEZONE']='1'

Usage: spark-submit [options] <app jar | python file | R file> [app arguments]
Usage: spark-submit --kill [submission ID] --master [spark://...]
Usage: spark-submit --status [submission ID] --master [spark://...]
Usage: spark-submit run-example [options] example-class [example args]
Options:
  --master MASTER_URL         spark://host:port, mesos://host:port, yarn,
                              k8s://https://host:port, or local (Default: local[*]).
用于指定master，如果是本地环境直接执行使用local[*]，否则根据具体使用的分布式调度系统来决定配置，如上所示，如果是yarn
则直接指定yarn即可，否则需要指定spark自带的分布式调度系统的ip地址和端口或k8s的ip地址和端口
  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or
                              on one of the worker machines inside the cluster ("cluster")
                              (Default: client).
执行模式，client或cluster模式
  --class CLASS_NAME          Your application's main class (for Java / Scala apps).
执行的main函数的名字，一般java中的主运行函数就是main
  --name NAME                 A name of your application.
执行的应用的名字可以自己自由定义
  --jars JARS                 Comma-separated list of jars to include on the driver
                              and executor classpaths.
spark运行的时候需要加载的额外的jar包，比如在pyspark的py文件中，引入分布式xgb和lgb需要配置其对应的jar包则需要从这里导入
  --packages                  Comma-separated list of maven coordinates of jars to include
                              on the driver and executor classpaths. Will search the local
                              maven repo, then maven central and any additional remote
                              repositories given by --repositories. The format for the
                              coordinates should be groupId:artifactId:version.
仍旧是引jar包，不过这里的jar包是直接从maven或gradle的中央仓库拉下来的，即运行的时候直接从远程下载，一般是下载下来
通过--jars来使用
  --exclude-packages          Comma-separated list of groupId:artifactId, to exclude while
                              resolving the dependencies provided in --packages to avoid
                              dependency conflicts.
为了避免package的命名冲突引入的参数，指--package的情况下，不引入哪些依赖
  --repositories              Comma-separated list of additional remote repositories to
                              search for the maven coordinates given with --packages.
选择中央仓库的地址，国内使用阿里云上的或其它国内云会快不少
  --py-files PY_FILES         Comma-separated list of .zip, .egg, or .py files to place
                              on the PYTHONPATH for Python apps.
spark-submit pyspark的py文件的应用的时候，通过这个参数指定py文件路径
  --files FILES               Comma-separated list of files to be placed in the working
                              directory of each executor. File paths of these files
                              in executors can be accessed via SparkFiles.get(fileName).
如果运行过程中需要使用到多个文件则可以使用上述的接口api来指定
  --conf PROP=VALUE           Arbitrary Spark configuration property.
--spark的conf设定，可以写多行，例如
    --conf spark.pyspark.python=ZIP_ENV/pyspark_py3/bin/python \
    --conf spark.pyspark.driver.python=/usr/bin/python \
    --conf spark.sql.catalogImplementation=hive \
  --properties-file FILE      Path to a file from which to load extra properties. If not
                              specified, this will look for conf/spark-defaults.conf.
--其它的一些preperties文件
  --driver-memory MEM         Memory for driver (e.g. 1000M, 2G) (Default: 1024M).
driver端的memory的内存资源申请配置
  --driver-java-options       Extra Java options to pass to the driver.
  --driver-library-path       Extra library path entries to pass to the driver.
  --driver-class-path         Extra class path entries to pass to the driver. Note that
                              jars added with --jars are automatically included in the
                              classpath.
--上述三个都是java相关的配置
  --executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).
--每个excutor端的内存资源申请配置
  --proxy-user NAME           User to impersonate when submitting the application.
                              This argument does not work with --principal / --keytab.
--执行任务的时候用户的名字
  --help, -h                  Show this help message and exit.
  --verbose, -v               Print additional debug output.
  --version,                  Print the version of current Spark.
 Cluster deploy mode only:
  --driver-cores NUM          Number of cores used by the driver, only in cluster mode
                              (Default: 1).
仅在cluster模式下才可以使用，即指定driver端使用的内核的数量
 Spark standalone or Mesos with cluster deploy mode only:
  --supervise                 If given, restarts the driver on failure.
  --kill SUBMISSION_ID        If given, kills the driver specified.
  --status SUBMISSION_ID      If given, requests the status of the driver specified.
 Spark standalone and Mesos only:
  --total-executor-cores NUM  Total cores for all executors.
 Spark standalone和mesos模式下使用的命令，没用过这两种模式
 Spark standalone and YARN only:
  --executor-cores NUM        Number of cores per executor. (Default: 1 in YARN mode,
                              or all available cores on the worker in standalone mode)
 Spark standalone 和 YARN 模式下使用的，用于指定每个excutor使用的内核的数量
 YARN-only:
  --queue QUEUE_NAME          The YARN queue to submit to (Default: "default").
  --num-executors NUM         Number of executors to launch (Default: 2).
                              If dynamic allocation is enabled, the initial number of
                              executors will be at least NUM.
  --archives ARCHIVES         Comma separated list of archives to be extracted into the
                              working directory of each executor.
这里需要详细展开说明以下，
在pyspark中，有三种参数
--archives，-files，py-files
--py-files PY_FILES:逗号隔开的的.zip、.egg、.py文件，这些文件会放置在PYTHONPATH下，该参数仅针对python应用程序
这里主要用于我们submit py文件的时候会import的其它的附属文件，这些文件会放置在PYTHONPATH下以便于submit执行py文件的时候
py文件能够import这些附属文件
注意submit的主py文件不可以通过py files设定，具体可见：
https://stackoverflow.com/questions/34209299/cannot-load-main-class-from-jar-file-in-spark-submit
--archives 主要用于定义python执行的env环境，env环境一般需要打包成zip文件。
--files FILES:逗号隔开的文件列表，这些文件将存放于每一个工作节点进程目录下
这里files和archives的作用基本类似：
https://www.codenong.com/38066318/ 下面解释部分来自于这个连接
通常，通过--files或--archives添加数据文件，并通过--py-files添加代码文件。
--py-files 将python文件 添加到类路径，以便您导入和使用。
http://spark.apache.org/docs/latest/programming-guide.html
Behind the scenes, pyspark invokes the more general spark-submit script.
You can add Python .zip, .egg or .py files to the runtime path by passing a comma-separated list to --py-files
http://spark.apache.org/docs/latest/running-on-yarn.html
The --files and --archives options support specifying file names with the # similar to Hadoop.
 For example you can specify: --files localtest.txt#appSees.txt and this will upload the file
 you have locally named localtest.txt into HDFS but this will be linked to by the name appSees.txt,
 and your application should use the name as appSees.txt to reference it when running on YARN.
http://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=addpyfile#pyspark.SparkContext.addPyFile
addFile(path) Add a file to be downloaded with this Spark job on every node. 
The path passed can be either a local file, a file in HDFS (or other Hadoop-supported filesystems),
 or an HTTP, HTTPS or FTP URI.
addPyFile(path) Add a .py or .zip dependency for
 all tasks to be executed on this SparkContext in the future.
 The path passed can be either a local file, a file in HDFS (or other Hadoop-supported filesystems),
 or an HTTP, HTTPS or FTP URI.
其中addfile和addpyfile是notebook或py文件中定义的，而不带add的是在命令行中定义的，在notebook或者py文件中
定义archives则通过spark.yarn.dist.archives(yarn模式下，其它模式没用过，暂时不知道怎么设置。。。)
https://spark.apache.org/docs/latest/api/python/user_guide/python_packaging.html?highlight=pyspark_python 详细可见，pyfile
原来也可以上传环境包的。。。
  --principal PRINCIPAL       Principal to be used to login to KDC, while running on
                              secure HDFS.
  --keytab KEYTAB             The full path to the file that contains the keytab for the
                              principal specified above. This keytab will be copied to
                              the node running the Application Master via the Secure
                              Distributed Cache, for renewing the login tickets and the
                              delegation tokens periodically.
上述都是yarn模式下专有的参数，--queue提交的队列名称  --num-executors 指定excutor的数量，--archives 重要，详细见上