小编eli*_*sah的帖子

去年我在iPhone开发者计划中收到了两个"技术支持事件",但是当我去续订时,我发现我会丢失它们.今年我还有另外两个,但我不确定使用它们可以解决哪些问题.

哪些问题最适合这些技术支持事件？我可以使用它们的问题是什么？你使用它们会得到什么？谢谢.

我想问一下,当我使用.join()时,你是否知道如何在pyspark中指定很多条件

示例:with hive:

query= "select a.NUMCNT,b.NUMCNT as RNUMCNT ,a.POLE,b.POLE as RPOLE,a.ACTIVITE,b.ACTIVITE as RACTIVITE FROM rapexp201412 b \
    join rapexp201412 a where (a.NUMCNT=b.NUMCNT and a.ACTIVITE = b.ACTIVITE and a.POLE =b.POLE  )\

但是在pyspark我不知道怎么做,因为以下内容:

df_rapexp201412.join(df_aeveh,df_rapexp2014.ACTIVITE==df_rapexp2014.ACTIVITE and df_rapexp2014.POLE==df_aeveh.POLE,'inner')

不起作用!!

我正在预测批量训练模型的流程之间的评级.我正在使用此处概述的方法:ALS模型 - 如何生成full_u*v ^ t*v？

! rm -rf ml-1m.zip ml-1m
! wget --quiet http://files.grouplens.org/datasets/movielens/ml-1m.zip
! unzip ml-1m.zip
! mv ml-1m/ratings.dat .

from pyspark.mllib.recommendation import Rating

ratingsRDD = sc.textFile('ratings.dat') \
               .map(lambda l: l.split("::")) \
               .map(lambda p: Rating(
                                  user = int(p[0]), 
                                  product = int(p[1]),
                                  rating = float(p[2]), 
                                  )).cache()

from pyspark.mllib.recommendation import ALS

rank = 50
numIterations = 20
lambdaParam = 0.1
model = ALS.train(ratingsRDD, rank, numIterations, lambdaParam)

然后提取产品功能......

import json
import numpy as np

pf = model.productFeatures()

pf_vals = pf.sortByKey().values().collect()
pf_keys …

我有一个数据帧,我想使用该数据帧的replace()函数之一 org.apache.spark.sql.DataFrameNaFunctions.

问题:我没有使用dataframe的实例获得智能(建议)中的这些方法.我明确地导入了这个类.

我无法找到任何可以给我演示如何使用这些函数或如何将数据帧转换为类型的东西DataFrameNaFunctions.

我尝试使用asInstanceof[]方法强制转换它但它抛出异常.

我试图理解Spark如何在引擎盖下改变依赖关系.因此我有两个问题:

1. 在Spark中,执行者如何知道从其他执行器获取数据？

   • 每个执行者在完成其地图侧任务后,是否将其状态和位置更新为某个中央实体(可能是驱动程序)并减少侧执行者首先联系驱动程序以获取每个执行程序的位置,然后直接从这些执行程序中取出？

2. 在具有随机播放依赖性的作业中,只有在所有地图侧任务完成后,驱动程序计划才会加入(或其他任务与shuffle依赖关系)？

   • 这是否意味着每个任务都会通知驱动程序其状态,驱动程序将及时协调其他相关任务.

我真的不喜欢用Caps Lock写的人.除了厌恶之外,它还污损了整个应用程序.我想知道如何防止用户使用大写锁定编写所有字符.由于特殊名称和缩写,我无法强制所有文本为小写.我应该使用什么逻辑？

我刚刚使用pip install elasticsearch安装了正确的软件包,但我的.py脚本找不到它.

我现在有这个:

ls /Library/Python/2.7/site-packages

README                       pip-1.5.6-py2.7.egg          urllib3-1.8.3-py2.7.egg-info virtualenv.py                virtualenv_support
easy-install.pth             urllib3                      virtualenv-1.11.6.dist-info  virtualenv.pyc

ls /usr/local/lib/python2.7/site-packages/

easy-install.pth              elasticsearch-1.0.0.dist-info setuptools-4.0.1-py2.7.egg    sitecustomize.py
elasticsearch                 pip-1.5.6-py2.7.egg           setuptools.pth                sitecustomize.pyc

现在当我运行我的脚本myelastic.py时:

import sys
print sys.path

from elasticsearch import Elasticsearch
es = Elasticsearch()

我有这个:

['/Users/tati/Desktop/python', '/Applications/MAMP/Library/lib/python27.zip', '/Applications/MAMP/Library/lib/python2.7', '/Applications/MAMP/Library/lib/python2.7/plat-darwin', '/Applications/MAMP/Library/lib/python2.7/plat-mac', '/Applications/MAMP/Library/lib/python2.7/plat-mac/lib-scriptpackages', '/Applications/MAMP/Library/lib/python2.7/lib-tk', '/Applications/MAMP/Library/lib/python2.7/lib-old', '/Applications/MAMP/Library/lib/python2.7/lib-dynload', '/Applications/MAMP/Library/lib/python2.7/site-packages']
Traceback (most recent call last):
  File "myelastic.py", line 5, in <module>
    from elasticsearch import Elasticsearch
ImportError: No module named elasticsearch

这是我第一次使用virtualenv,但我不知道如何解决这个问题,谢谢!

我有一个CSV文件,格式如下:

product_id1,product_title1
product_id2,product_title2
product_id3,product_title3
product_id4,product_title4
product_id5,product_title5
[...]

product_idX是一个整数,product_titleX是一个String,例如:

453478692, Apple iPhone 4 8Go

我正在尝试从我的文件创建TF-IDF,所以我可以将它用于MLlib中的朴素贝叶斯分类器.

到目前为止,我正在使用Spark for Scala并使用我在官方页面和Berkley AmpCamp 3和4上找到的教程.

所以我正在读文件:

val file = sc.textFile("offers.csv")

然后我将它映射到元组中 RDD[Array[String]]

val tuples = file.map(line => line.split(",")).cache

在我将元组转换成对之后 RDD[(Int, String)]

val pairs = tuples.(line => (line(0),line(1)))

但我被困在这里,我不知道如何从它创建Vector,把它变成TFIDF.

谢谢

我正在优化Spark中的参数,并且想知道Spark是如何改组数据的.

确切地说,我有一个简单的字数统计程序,并想知道spark.shuffle.file.buffer.kb如何影响运行时间.现在,当我将此参数设置得非常高时,我只看到减速(我猜这会阻止每个任务的缓冲区同时适应内存).

有人可以解释Spark是如何进行减少的吗？例如,在RDD中读取和分区数据,并且当调用"action"函数时,Spark会将任务发送到工作节点.如果操作是减少,Spark如何处理此问题,以及与此过程相关的shuffle文件/缓冲区如何？

我正在使用RandomForest.featureImportances但我不理解输出结果.

我有12个功能,这是我得到的输出.

我知道这可能不是一个特定于apache-spark的问题,但我无法找到解释输出的任何地方.

// org.apache.spark.mllib.linalg.Vector = (12,[0,1,2,3,4,5,6,7,8,9,10,11],
 [0.1956128039688559,0.06863606797951556,0.11302128590305296,0.091986700351889,0.03430651625283274,0.05975817050022879,0.06929766152519388,0.052654922125615934,0.06437052114945474,0.1601713590349946,0.0324327322375338,0.057751258970832206])