小编fem*_*yte的帖子

有人可以向我解释以下内容:

scala> def squared(x: Int) = x * x
squared: (x: Int)Int
scala> val sq : (Int) => Int = squared
sq: Int => Int = <function1>
scala>  sq.getClass
res111: Class[_ <: Int => Int] = class $anonfun$1

到目前为止我理解这一点,sqared是一个函数,而sq是一个函数指针.

但后来我这样做:

scala> squared.getClass
<console>:13: error: missing arguments for method squared;
follow this method with `_' if you want to treat it as a partially applied function
   squared.getClass
   ^

为什么我不能在平方上调用getClass？毕竟,不是函数的第一类对象吗？为什么我需要这样做呢？

scala> squared(7).getClass
res113: Class[Int] = int

我也得到了相同的结果

scala> sq(5).getClass
res115: Class[Int] = …

对于dask DataFrame,pandas中的sort_values相当于什么？我正在尝试扩展一些具有内存问题的Pandas代码,而不是使用dask DataFrame.

相当于:

ddf.set_index([col1, col2], sorted=True)

？

我写了一些 python 代码，大量使用了 Pandas 库。代码似乎有点慢，所以我通过 cProfile 运行它以查看瓶颈在哪里。根据 cProfile 结果，瓶颈之一是对 pandas.lib_scalar_compare 的调用：

1604  262.301    0.164  262.301    0.164 {pandas.lib.scalar_compare}

我的问题是 - 在什么情况下会调用它？当我选择 DataFrame 的一部分时，我假设它。这是我的代码的样子：

if (var=='9999'):
    dataTable=resultTable.ix[(resultTable['col1'] == var1)  
                                             & (resultTable['col2']==var2)].copy() 
else:
    dataTable=resultTable.ix[(resultTable['col1'] == var1)  
                                           & (resultTable['col2']==var2)
                                           & (resultTable['col3']==int(val3))].copy() 

我有以下问题：

1. 那是最终调用导致瓶颈的代码的代码片段吗？
2. 如果是这样，无论如何要优化它？我目前使用的熊猫版本是pandas-0.8。

对此的任何帮助将不胜感激。

有没有办法暂停Dataproc集群,所以当我没有主动运行spark-shell或spark-submit工作时,我不会收到账单？此链接上的群集管理说明:https://cloud.google.com/sdk/gcloud/reference/beta/dataproc/clusters/

仅显示如何销毁群集,但我已安装spark cassandra连接器API.我只是创建一个每次都需要安装的图像的唯一选择吗？

我有一组日志文件,我想读入RDD.这些日志文件都是压缩的gzip文件,文件名是日期戳.

我一直在sc.wholeTextFiles()阅读文件,似乎我一直在遇到Java堆内存问题.为了隔离问题,我决定在一台机器上针对单个文件运行它作为测试用例.

我从这里获得了这个文件:

http://dumps.wikimedia.org/other/pagecounts-raw/

以下是文件的大小,包括压缩和未压缩版本:

 myuser@fembuntu$ ls -ltr pagecounts-20090505-180000*
 -rw-rw-r-- 1 myuser myuser  65170192 Sep 20  2009 pagecounts-20090505-180000.gz
-rw-rw-r-- 1 myuser myuser 233007266 Jan 22 00:23 pagecounts-20090505-180000.txt

并且机器上的可用内存如下:

myuser@fembuntu:~$ free -tm

       total       used       free     shared    buffers     cached
Mem:    4856       3018       1838        123         27        407
-/+ buffers/cache: 2583       2273
 Swap:  5080        849       4231
Total:  9937       3867       6069

所以我启动了spark-shell,给执行程序2G内存:

$ spark-shell --executor-memory 2G

scala> val pc_loc = "file:///home/myuser/data/pagecounts"
scala> val filename="/pagecounts-20090505-180000.gz"
filename: String = /pagecounts-20090505-180000.gz

在这里,我通过数据读取sc.textFile()并显示前2行:

scala>  var …

我有以下问题 - 我可以使用一个参数的值来定义另一个参数吗？这是我正在尝试做的事情的说明。假设我有一个如下所示的配置文件：

[MyTaskRunner]
logdir=/tmp/logs
numruns=2

我MyTaskRunner这样定义：

class MyTaskRunner(luigi.Task):
      logdir=luigi.Parameter(default=None)
      rundate=luigi.Parameter(default=today)

其中logdir是从配置文件中获取的参数，rundate是运行时传入的参数。

现在，假设我想定义一个logpath_str像这样的新变量

       logpath_str="{}/{}".format(logdir, rundate)

是否可以将其定义为参数？

解决方案是指定默认值，如下所示：

       logpath=luigi.Parameter(default=logpath_str)

欢迎任何建议。

我定义了以下元组列表:

var temps = Seq(("Spain", Seq(68,70,73,75)),       
                ("Trinidad",Seq(87,83,88,84,88)), 
                ("England",Seq(52,55,58,57.5)),
                ("Eritrea",Seq(90,91.3,88,91)))

结果类型如下:

temps: Seq[(String, Seq[AnyVal])] = List((Spain,List(68, 70, 73, 75)), (Trinidad,List(87, 83, 88, 84, 88)), (England,List(52.0, 55.0, 58.0, 57.5)), (Eritrea,List(90.0, 91.3, 88.0, 91.0)))

我的问题是为什么temps的类型被推断为Seq[(String, Seq[AnyVal])]而不是Seq[(String, Seq[Int])] or Seq[(String, Seq[Double])]人们所期望的？

如果有一个简单的解释,我是斯卡拉新手,请原谅我.

我正在使用 Dataproc 通过 Spark-shell 在集群上运行 Spark 命令。我经常收到错误/警告消息，表明我与执行者失去了连接。这些消息看起来像这样：

[Stage 6:>                                                          (0 + 2) / 2]16/01/20 10:10:24 ERROR     org.apache.spark.scheduler.cluster.YarnScheduler: Lost executor 5 on spark-cluster-femibyte-w-0.c.gcebook-1039.internal: remote Rpc client disassociated
16/01/20 10:10:24 WARN akka.remote.ReliableDeliverySupervisor:  Association with remote system [akka.tcp://sparkExecutor@spark-cluster-  femibyte-w-0.c.gcebook-1039.internal:60599] has failed, address is now gated for [5000] ms. Reason: [Disassociated] 
16/01/20 10:10:24 WARN org.apache.spark.scheduler.TaskSetManager: Lost task 0.2 in stage 6.0 (TID 17, spark-cluster-femibyte-w-0.c.gcebook-1039.internal): ExecutorLostFailure (executor 5 lost)
16/01/20 10:10:24 WARN org.apache.spark.scheduler.TaskSetManager: Lost task 1.2 in stage 6.0 (TID 16, spark-cluster-femibyte-w-0.c.gcebook-1039.internal): ExecutorLostFailure (executor 5 …

我使用以下代码为我从MySQL导入的文件创建一个RDD,使用Sqoop进入Hive:

def rddFromParquetHdfsFile(path: String): RDD[GenericRecord] = {
val job = new Job()
FileInputFormat.setInputPaths(job, path)
ParquetInputFormat.setReadSupportClass(job,
    classOf[AvroReadSupport[GenericRecord]])
return sc.newAPIHadoopRDD(job.getConfiguration,
    classOf[ParquetInputFormat[GenericRecord]],
    classOf[Void],
    classOf[GenericRecord]).map(x => x._2)
}
val warehouse = "hdfs://quickstart/user/hive/warehouse/"
val order_items = rddFromParquetHdfsFile(warehouse + "order_items");
val products = rddFromParquetHdfsFile(warehouse + "products");

我现在尝试查看前5个产品:

products.take(5)

我最终得到以下错误:

org.apache.spark.SparkException: 
Job aborted due to stage failure: Task 0.0 in stage 0.0 (TID 0) had a 
not serializable result: org.apache.avro.generic.GenericData$Record
Serialization stack:
- object not serializable (class:   
  org.apache.avro.generic.GenericData$Record, value: {"product_id": 1,  
  "product_category_id": 2, "product_name": "Quest Q64 10 …

我对Scala REPL中的这种行为感到很困惑:

scala> import java.util.Map
import java.util.Map

scala> import java.util.HashMap
import java.util.HashMap

scala> val jMap:java.util.Map[String,Int]=new HashMap[String,Int]("first"->1,"second" -> 2)

<console>:12: error: type mismatch;
found   : (String, Int)
required: Float
   val jMap =new HashMap[String,Int]("first"->1,"second" -> 2)
                                                                     ^
<console>:12: error: type mismatch;
found   : (String, Int)
required: Float
   val jMap=new HashMap[String,Int]("first"->1,"second" -> 2)
                                                                                  ^

有人可以帮助解释这里发生了什么吗？

我是 Kubernetes 的新手，我正在学习它。是否有使用 Kubernetes 创建 Docker 映像而不是从存储库中提取它的情况？

任何人都可以向我解释为什么这个条目在我输入Scala REPL时会杀死编译器？

def combination(x: Int, y: Int, f: (Int,Int) => Int) = f(x,y)

我正在使用Scala 2.11.5

    at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:904)
    at scala.reflect.internal.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:97)
    at scala.tools.nsc.interpreter.ILoop.process(ILoop.scala:904)
    at scala.tools.nsc.MainGenericRunner.runTarget$1(MainGenericRunner.scala:74)
    at scala.tools.nsc.MainGenericRunner.run$1(MainGenericRunner.scala:87)
    at scala.tools.nsc.MainGenericRunner.process(MainGenericRunner.scala:98)
    at scala.tools.nsc.MainGenericRunner$.main(MainGenericRunner.scala:103)
    at scala.tools.nsc.MainGenericRunner.main(MainGenericRunner.scala)

That entry seems to have slain the compiler.  
Shall I replay your session? I can re-run each line except the last one.