pyspark 使用時環境設置

 

在腳本中導入pyspark的流程

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

 

設置pyspark運行時的python版本

vi ~/.bashrc

export PYSPARK_PYTHON=/usr/local/bin/python3 

export PYSPARK_DRIVER_PYTHON=ipython3

編輯完保存退出

source ~/.bashrc

 

使用pyspark處理hbase缺乏jar包時需配置環境

spark加載配置的默認目錄是 SPARK_HOME/conf/spark-env.sh ,不存在此目錄此文件時可自行建立

通常來講在spark-env.sh的末尾須要添加幾行

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)   不添加這一行可能致使java class not found 之類的異常

export JAVA_HOME=/usr/java/jdk1.8.0_191-amd64/jre

export HADOOP_HOME=/usr/local/hadoop

export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop

export SPARK_MASTER_HOST=HDP-master

export SPARK_WORKER_CORES=4     設置每一個worker最多使用的核數，可設置爲機器的內核數

export SPARK_WORKER_MEMORY=4g    設置每一個worker最多使用的內存

 

spark處理hbase時須要一些hbase的jar包，能夠在SPARK_HOME/jars/下新建一個hbase目錄，而後將HBASE_HOME/lib/下面的相關包都複製過來

（也可單獨複製lib目錄下的這些包 hbase*.jar ,guava-12.0.1.jar,htrace-core-3.1.0-incubating.jar , protobuf-java-2.5.0.jar ）

另外需下載把hbase的數據轉換爲Python可讀取的jar包 spark-example-1.6.0.jar

(下載頁面地址爲https://mvnrepository.com/artifact/org.apache.spark/spark-example_2.11/1.6.0-typesafe-001 )

這樣就須要將spark-env.sh中的SPARK_DIST_CLASSPATH的值修改成

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath):$(/usr/local/hbase/bin/hbase classpath):/usr/local/spark/jars/hbase/*

 

使用spark讀寫hbase的相關代碼流程

host = 'master,slave1,slave2'

hbase_table = 'TEST:test1'

conf = {"hbase.zookeeper.quorum":host,"hbase.mapreduce.inputtable":hbase_table}

keyConv = "org.apache.spark.examples.pythonconverters.ImmutableBytesWritableToStringConverter"

valueConv = "org.apache.spark.examples.pythonconverters.HBaseResultToStringConverter"

# 讀取habse表中的數據到rdd

hbase_rdd = sc.newAPIHadoopRDD("org.apache.hadoop.hbase.mapreduce.TableInputFormat","org.apache.hadoop.hbase.io.ImmutableBytesWritable",

"org.apache.hadoop.hbase.client.Result",keyConverter=keyConv,valueConverter=valueConv,conf=conf)

count = hbase_rdd.count()

one = hbase_rdd.first()            查看rdd的第一條數據tuple(rowkey,'\n'.join(str(json_value)))

one_value = one[1].split('\n')

one_value[1]    形式爲'{"qualifier":"列名","timestamp":"1560533059864","columnFamily":"列簇名", "row":"0000632232_1550712079","type":"Put","value":"0"}'

       

寫入hbase

write_table = 'student'

write_keyConv = "org.apache.spark.examples.pythonconverters.StringToImmutableBytesWritableConverter"

write_valueConv= "org.apache.spark.examples.pythonconverters.StringListToPutConverter"

conf = {"hbase.zookeeper.quorum":host,"hbase.mapred.outputtable":table,"mapreduce.outputformat.class":"org.apache.hadoop.hbase.mapreduce.TableOutputFormat",

"mapreduce.job.output.key.class":"org.apache.hadoop.habse.io.ImmutableBytesWritable","mapreduce.job.output.value.class":"org.apache.hadoop.io.Writable"}

rawData = ['3,info,age,19','4,info,age,17'] # 最後將數據改爲[rowkey,[rowkey,column family, column name,value]]形式寫進hbase

sc.parallelize(rawData).map(lambda x:(x[0],x.split(','))).saveAsNewAPIHadoopDataset(conf=conf,keyConverter=keyConv,valueConverter=valueConv) 

  

 

 

 

spark啓動後對應的進程是WORKER 和 MASTER