pyspark kütüphanesi pyspark'ta nasıl edinilir?

Question

pyspark'ta matplotlib.bblpath veya shapely.geometry kitaplıklarını kullanmak istiyorum.

ben hiçbirini aşağıda hatayı alıyorum almaya çalıştığınızda:

>>> from shapely.geometry import polygon 
Traceback (most recent call last): 
File "<stdin>", line 1, in <module> 
ImportError: No module named shapely.geometry 

Ben modül mevcut değil biliyorum ama bu paketler benim pyspark kütüphaneleri getirilebilir nasıl bilmek istiyorum.

SparkContext.addPyFile("module.py") # also .zip 

, docs alıntı:

Answer 1

Kıvılcım bağlamda kullanmayı deneyin

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.

Answer 2

bu bağımsız üzerinde (yani dizüstü/masaüstü) mı yoksa bir küme ortamında (örn AWS EMR)? dizüstü/masaüstü Eğer

1. , pip install shapely sadece iyi çalışması gerekir. Varsayılan python ortamınız için ortam değişkenlerinizi kontrol etmeniz gerekebilir. Örneğin, tipik olarak Python 3'ü kullanır ancak pyspark için Python 2'yi kullanırsanız, pyspark için düzgün bir şekilde mevcut olmazdı. Böyle AWS EMR gibi bir küme ortamında ise

2. , deneyebilirsin:

   import os 
   
   def myfun(x):` 
        os.system("pip install shapely") 
        return x 
   rdd = sc.parallelize([1,2,3,4]) ## assuming 4 worker nodes 
   rdd.map(lambda x: myfun(x)).collect() 
   ## call each cluster to run the code to import the library 
   

"Ben modül mevcut değil biliyorum ama bu paketler olabilir nasıl bilmek istiyorum pyspark kütüphanemize getirildi.

EMR'de, pyspark'ın istediğiniz diğer kütüphane ve konfigürasyonlarla önceden hazırlanmasını istiyorsanız, bu ayarları yapmak için bir önyükleme adımı kullanabilirsiniz. Bunun yanı sıra, Scala'da Spark'i derlemeden pyspark'a bir kitaplık ekleyemezsiniz (SBT ile ilgili bilgi sahibi değilseniz, yapmanız gereken bir acı olacaktır).

Answer 3

AWS EMR kümemizde bu şekilde çalışıyorum (Başka kümelerde de aynı şekilde olmalıdır). Aşağıdaki shell script oluşturulan ve bir önyükleme-eylemler olarak infaz:

#!/bin/bash 
# shapely installation 
wget http://download.osgeo.org/geos/geos-3.5.0.tar.bz2 
tar jxf geos-3.5.0.tar.bz2 
cd geos-3.5.0 && ./configure --prefix=$HOME/geos-bin && make && make install 
sudo cp /home/hadoop/geos-bin/lib/* /usr/lib 
sudo /bin/sh -c 'echo "/usr/lib" >> /etc/ld.so.conf' 
sudo /bin/sh -c 'echo "/usr/lib/local" >> /etc/ld.so.conf' 
sudo /sbin/ldconfig 
sudo /bin/sh -c 'echo -e "\nexport LD_LIBRARY_PATH=/usr/lib" >> /home/hadoop/.bashrc' 
source /home/hadoop/.bashrc 
sudo pip install shapely 
echo "Shapely installation complete" 
pip install https://pypi.python.org/packages/74/84/fa80c5e92854c7456b591f6e797c5be18315994afd3ef16a58694e1b5eb1/Geohash-1.0.tar.gz 
# 
exit 0 

Not: Yerine bu komut bir kümedeki her düğüm bağımsız çalıştırılabilir bir önyükleme-eylemler olarak çalışan. Her iki senaryoyu da test ettim.

Python 2.7.10 (default, Dec 8 2015, 18:25:23) 
[GCC 4.8.3 20140911 (Red Hat 4.8.3-9)] on linux2 
Type "help", "copyright", "credits" or "license" for more information. 
Welcome to 
     ____    __ 
    /__/__ ___ _____/ /__ 
    _\ \/ _ \/ _ `/ __/ '_/ 
    /__/.__/\_,_/_/ /_/\_\ version 1.6.1 
     /_/ 

Using Python version 2.7.10 (default, Dec 8 2015 18:25:23) 
SparkContext available as sc, HiveContext available as sqlContext. 
>>> from pyspark.sql.functions import udf 
>>> from pyspark.sql.types import StringType 
>>> from shapely.wkt import loads as load_wkt 
>>> def parse_region(region): 
...  from shapely.wkt import loads as load_wkt 
...  reverse_coordinate = lambda coord: ' '.join(reversed(coord.split(':'))) 
...  coordinate_list = map(reverse_coordinate, region.split(', ')) 
...  if coordinate_list[0] != coordinate_list[-1]: 
...   coordinate_list.append(coordinate_list[0]) 
...  return str(load_wkt('POLYGON ((%s))' % ','.join(coordinate_list)).wkt) 
... 
>>> udf_parse_region=udf(parse_region, StringType()) 
16/09/06 22:18:34 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 1.2.0 
16/09/06 22:18:34 WARN ObjectStore: Failed to get database default, returning NoSuchObjectException 
>>> df = sqlContext.sql('select id, bounds from <schema.table_name> limit 10') 
>>> df2 = df.withColumn('bounds1', udf_parse_region('bounds')) 
>>> df2.first() 
Row(id=u'0089d43a-1b42-4fba-80d6-dda2552ee08e', bounds=u'33.42838509594465:-119.0533447265625, 33.39170168789402:-119.0203857421875, 33.29992542601392:-119.0478515625', bounds1=u'POLYGON ((-119.0533447265625 33.42838509594465, -119.0203857421875 33.39170168789402, -119.0478515625 33.29992542601392, -119.0533447265625 33.42838509594465))') 
>>> 

Teşekkür, Hüseyin Bohra