小编Bam*_*mqf的帖子

在pyspark 1.6.2中,我可以导入col函数

from pyspark.sql.functions import col

但是当我尝试在Github源代码中查找它时,我发现文件中没有col函数,functions.pypython如何导入一个不存在的函数？

作为简化示例,我尝试使用以下代码过滤Spark DataFrame:

val xdf = sqlContext.createDataFrame(Seq(
  ("A", 1), ("B", 2), ("C", 3)
)).toDF("name", "cnt")
xdf.filter($"cnt" >1 || $"name" isin ("A","B")).show()

然后它错误:

org.apache.spark.sql.AnalysisException: cannot resolve '((cnt > 1) || name)' due to data type mismatch: differing types in '((cnt > 1) || name)' (boolean and string).;

什么是正确的方法呢？在我看来,它在name列之后停止阅读.它是解析器中的错误吗？我正在使用Spark 1.5.1

我尝试按照NERDTree github网站上的FAQ部分中的说明操作:

"问:如何在vim启动时自动打开NERDTree？"

"A.把它粘在你的vimrc中:autocmd vimenter * NERDTree"

它工作但是当我打开文件时光标停留在NEARDTree资源管理器区域但不在编辑区域中,我必须按下Ctrl+w+l将其移回,我应该在.vimrc文件中写什么来自动在编辑区域中设置光标？

我正在尝试关于Typescript的角度2 的官方5 Min快速入门.我按照所有步骤操作,当我运行npm start它时,我My First Angular 2 App在firefox浏览器上弹出了预期的消息.

但是,当我检查控制台时,我发现错误消息:

Error: watch node_modules/angular2/es6/dev/src/common/directives/ng_if.js ENOSPC

我不知道它是否与我之前在npm install项目的根文件夹上运行时收到的警告消息有关:

npm WARN install Couldn't install optional dependency: Unsupported
npm WARN EPACKAGEJSON angular2-quickstart@1.0.0 No description
npm WARN EPACKAGEJSON angular2-quickstart@1.0.0 No repository field.

有人能告诉我我的方法有什么问题吗？更详细的控制台输出如下所示:

> angular2-quickstart@1.0.0 start /home/ubuntu/jsProjects/angular2-quickstart
> concurrent "npm run tsc:w" "npm run lite" 

[0] 
[0] > angular2-quickstart@1.0.0 tsc:w /home/ubuntu/jsProjects/angular2-quickstart
[0] > tsc -w
[0] 
[1] 
[1] > angular2-quickstart@1.0.0 lite /home/ubuntu/jsProjects/angular2-quickstart
[1] > lite-server
[1] 
[1] [BS] …

假设我有一个csv文件,如下所示:

Type,ID,NAME,CONTENT,RESPONSE,GRADE,SOURCE
A,3,"","I have comma, ha!",I have open double quotes",A,""

期望的输出应该是:

df <- data.frame(Type='A',ID=3, NAME=NA, CONTENT='I have comma, ha!',
                 RESPONSE='I have open double quotes\"', GRADE=A, SOURCE=NA)
df
  Type ID NAME           CONTENT                   RESPONSE GRADE SOURCE
1    A  3   NA I have comma, ha! I have open double quotes"     A     NA

我尝试使用read.csv,因为数据提供程序使用quote来转义字符串中的逗号,但是他们忘记了在没有逗号的字符串中转义双引号,所以无论我是否禁用引用,read.csv我都不会获得所需的输出.

我怎么能在R中这样做？其他包装解决方案也欢迎.

我需要从互联网下载一个压缩的 csv 文件，解压缩它并将其作为 data.frame 加载到内存中。有没有办法只在内存中完成整个过程而无需读/写磁盘？这是我尝试过的：

library(RCurl)
file <- getURL(url, userpwd='user:pwd')
tf <- tempfile()
writeBin(content(file, "raw"), tf)
tf <- unzip(tf, exdir=tempdir())
data <- read.csv(tf)
unlink(tf)

我想writeBin，unzip和read.csv这里使用昂贵的磁盘I / O，但我不知道如何改进它。

我试图在scala sbt应用程序中使用spark而不是spark-submit直接使用.

我已经有一个远程纱线集群正在运行,我可以连接到SparkR中的纱线集群运行火花作业.但是当我尝试在scala应用程序中执行类似的操作时,它无法将我的环境变量加载到yarn配置中,而是使用默认的yarn地址和端口.

sbt应用程序只是一个简单的对象:

object simpleSparkApp {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("simpleSparkApp")
      .setMaster("yarn-client")
      .set("SPARK_HOME", "/opt/spark-1.5.1-bin-hadoop2.6")
      .set("HADOOP_HOME", "/opt/hadoop-2.6.0")
      .set("HADOOP_CONF_DIR", "/opt/hadoop-2.6.0/etc/hadoop")
    val sc = new SparkContext(conf)
  }
}

当我在Intellij IDEA中运行此应用程序时,日志显示:

15/11/15 18:46:05 INFO RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/11/15 18:46:06 INFO Client: Retrying connect to server: 0.0.0.0/0.0.0.0:8032. Already tried 0 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1 SECONDS)
15/11/15 18:46:07 INFO Client: Retrying connect to server: 0.0.0.0/0.0.0.0:8032. Already tried 1 time(s); retry policy …

我在hadoop集群中有一个带有重复列的10GB csv文件.我尝试在SparkR中分析它,所以我使用spark-csv包来解析它DataFrame:

  df <- read.df(
    sqlContext,
    FILE_PATH,
    source = "com.databricks.spark.csv",
    header = "true",
    mode = "DROPMALFORMED"
  )

但由于df有重复的Email列,如果我想选择此列,则会出错:

select(df, 'Email')

15/11/19 15:41:58 ERROR RBackendHandler: select on 1422 failed
Error in invokeJava(isStatic = FALSE, objId$id, methodName, ...) : 
  org.apache.spark.sql.AnalysisException: Reference 'Email' is ambiguous, could be: Email#350, Email#361.;
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolve(LogicalPlan.scala:278)
...

我想保留第一次出现的Email列并删除后者,我该怎么做？

我想过滤掉Spark DataFrame中具有看似真实的电子邮件列的行,这是我尝试过的:

df.filter($"Email" match {case ".*@.*".r => true case _ => false})

但这不起作用.做正确的方法是什么？

我在HDFS上有许多镶木地板文件目录,每个目录包含几千个小的(大多数<100kb)镶木地板文件.他们放慢了我的Spark工作,所以我想把它们结合起来.

使用以下代码,我可以将本地镶木地板文件重新分区为较少的部分:

val pqFile = sqlContext.read.parquet("file:/home/hadoop/data/file.parquet")
pqFile.coalesce(4).write.save("file:/home/hadoop/data/fileSmaller.parquet")

但我不知道如何通过编程方式通过Scala代码获取HDFS上的目录大小,因此我无法计算出传递给coalesce实际数据集的函数的分区数.

我怎样才能做到这一点？或者在Spark中有一个方便的方法,以便我可以配置编写器来编写固定大小的镶木地板分区？