Pyspark integration into Pycharm -

-

i'm bit lost on how confiure pycharm can directly run scripts within pyspark. i'm using pyspark ontop of elasticsearch cluster , using following code run script. it's running default python interpreter tried configure pyspark shell interpreter, didn't work error it's not valid sdk:

__author__ = 'lucas'   pyspark import sparkcontext, sparkconf  if __name__ == "__main__":      conf = sparkconf().setappname("estest")     sc = sparkcontext(conf=conf)      es_read_conf = {         "es.nodes" : "localhost",         "es.port" : "9200",         "es.resource" : "titanic/passenger"     }     es_rdd = sc.newapihadooprdd(         inputformatclass="org.elasticsearch.hadoop.mr.esinputformat",         keyclass="org.apache.hadoop.io.nullwritable",         valueclass="org.elasticsearch.hadoop.mr.linkedmapwritable",         conf=es_read_conf)      es_write_conf = {         "es.nodes" : "localhost",         "es.port" : "9200",         "es.resource" : "titanic/value_counts"     }      doc = es_rdd.first()[1]      field in doc:          value_counts = es_rdd.map(lambda item: item[1][field])         value_counts = value_counts.map(lambda word: (word, 1))         value_counts = value_counts.reducebykey(lambda a, b: a+b)         value_counts = value_counts.filter(lambda item: item[1] > 1)         value_counts = value_counts.map(lambda item: ('key', {             'field': field,             'val': item[0],             'count': item[1]         }))          value_counts.saveasnewapihadoopfile(             path='-',             outputformatclass="org.elasticsearch.hadoop.mr.esoutputformat",             keyclass="org.apache.hadoop.io.nullwritable",             valueclass="org.elasticsearch.hadoop.mr.linkedmapwritable",             conf=es_write_conf)

but generates following stacktrace:

traceback (most recent call last):   file "/home/lucas/pycharmprojects/tweetspark/analytics/tweetanalyzer.py", line 20, in <module>     conf=es_read_conf)   file "/var/opt/spark/python/pyspark/context.py", line 601, in newapihadooprdd     jconf, batchsize)   file "/var/opt/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__   file "/var/opt/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value py4j.protocol.py4jjavaerror: error occurred while calling z:org.apache.spark.api.python.pythonrdd.newapihadooprdd. : java.lang.classnotfoundexception: org.elasticsearch.hadoop.mr.linkedmapwritable     @ java.net.urlclassloader$1.run(urlclassloader.java:366)     @ java.net.urlclassloader$1.run(urlclassloader.java:355)     @ java.security.accesscontroller.doprivileged(native method)     @ java.net.urlclassloader.findclass(urlclassloader.java:354)     @ java.lang.classloader.loadclass(classloader.java:425)     @ java.lang.classloader.loadclass(classloader.java:358)     @ java.lang.class.forname0(native method)     @ java.lang.class.forname(class.java:278)     @ org.apache.spark.util.utils$.classforname(utils.scala:179)     @ org.apache.spark.api.python.pythonrdd$.newapihadooprddfromclassnames(pythonrdd.scala:519)     @ org.apache.spark.api.python.pythonrdd$.newapihadooprdd(pythonrdd.scala:503)     @ org.apache.spark.api.python.pythonrdd.newapihadooprdd(pythonrdd.scala)     @ sun.reflect.nativemethodaccessorimpl.invoke0(native method)     @ sun.reflect.nativemethodaccessorimpl.invoke(nativemethodaccessorimpl.java:57)     @ sun.reflect.delegatingmethodaccessorimpl.invoke(delegatingmethodaccessorimpl.java:43)     @ java.lang.reflect.method.invoke(method.java:606)     @ py4j.reflection.methodinvoker.invoke(methodinvoker.java:231)     @ py4j.reflection.reflectionengine.invoke(reflectionengine.java:379)     @ py4j.gateway.invoke(gateway.java:259)     @ py4j.commands.abstractcommand.invokemethod(abstractcommand.java:133)     @ py4j.commands.callcommand.execute(callcommand.java:79)     @ py4j.gatewayconnection.run(gatewayconnection.java:207)     @ java.lang.thread.run(thread.java:745)

what missing elasticsearch-spark.jar. download elasticsearch-hadoop integration, find elasticsearch-spark under dist subdirectory, set spark_classpath environment variable 

os.environ['spark_classpath'] = "/path/to/elasticsearch-hadoop-2.3.0/dist/elasticsearch-spark_2.10-2.3.0.jar"

Comments

Post a Comment

Popular posts from this blog

python - Healpy: From Data to Healpix map -

-
Image
i have data grid rows represent theta (0, pi) , columns represent phi (0, 2*pi) , f(theta,phi) density of dark matter @ location. wanted calculate power spectrum , have decided use healpy. what can not understand how format data healpy use. if provide code (in python obvious reasons) or point me tutorial, great! have tried hand @ doing following code: #grid dimensions nrows*ncols (subject change) theta = np.linspace(0, np.pi, num=grid.shape[0])[:, none] phi = np.linspace(0, 2*np.pi, num=grid.shape[1]) nside = 512 print "pixel area: %.2f square degrees" % hp.nside2pixarea(nside, degrees=true) pix = hp.ang2pix(nside, theta, phi) healpix_map = np.zeros(hp.nside2npix(nside), dtype=np.double) healpix_map[pix] = grid but, when try execute code power spectrum. specifically, : cl = hp.anafast(healpix_map[pix], lmax=1024) i error: typeerror: bad number of pixels if point me tutorial or edit code great. more specifications: data in 2d np array , can change numrows/...
Read more

c - Bitwise operation with (signed) enum value -

-
i using enumerator values flags: typedef enum { = 0x00, b = 0x01u, // u has no influence, expected c = 0x02u, // u has no influence, expected ... } enum_name; volatile unsigned char* reg = someaddress; *reg |= b; according misra-c:2004 bitwise operations shall not done signed type. unfortunately, compiler iar use signed int (or short or char) underlying type of enums, , option can find relates size, not signedness ("--enum-is-int"). according iar c/c++ development guide arm , pages 169 , 211, can define type of enum s if enable iar language extensions ( -e command-line option, or project > options > c/c++ compiler > language > allow iar extensions in ide). in particular, should define "sentinel" value, make sure compiler chooses correct type. prefers signed types, , uses smallest possible integer type, sentinel should largest positive integer corresponding unsigned integer type can describe. example, typedef enum ...
Read more

xslt - Unnest parent nodes by child node -

-
in xml there items 0-n attributes , item should copied each attribute new item 1 attribute. given xml this: <?xml version="1.0" encoding="utf-8" ?> <items> <item> <name>a</name> <attributes> <attribute> <key>attribute1</key> <value>1</value> </attribute> </attributes> </item> <item> <name>b</name> </item> <item> <name>c</name> <attributes> <attribute> <key>attribute1</key> <value>5</value> </attribute> <attribute> <key>attribute2</key> <value>2</value> </attribute> <attribute> <key>attri...
Read more