我只是想知道Apache Spark中的RDD和DataFrame (Spark 2.0.0 DataFrame只是一个类型别名Dataset[Row])有什么区别？
你能把一个转换成另一个吗？
当我尝试理解CAP中的"可用性"(A)和"分区容差"(P)时,我发现很难理解各篇文章中的解释.
我觉得A和P可以在一起(我知道事实并非如此,这就是我无法理解的原因!).
用简单的术语解释,A和P是什么以及它们之间的区别？
我似乎无法在MongoDB中使用最基本的日期查询.使用看起来像这样的文档:
{
"_id" : "foobar/201310",
"ap" : "foobar",
"dt" : ISODate("2013-10-01T00:00:00.000Z"),
"tl" : 375439
}
一个看起来像这样的查询:
{
"dt" : {
"$gte" : {
"$date" : "2013-10-01T00:00:00.000Z"
}
}
}
我从执行中得到0结果:
db.mycollection.find({
"dt" : { "$gte" : { "$date" : "2013-10-01T00:00:00.000Z"}}
})
知道为什么这不起作用吗？
作为参考,此查询由Spring的MongoTemplate生成,因此我无法直接控制最终发送到MongoDB的查询.
(PS)
> db.version()
2.4.7
谢谢!
如何将RDD(org.apache.spark.rdd.RDD[org.apache.spark.sql.Row])转换为Dataframe org.apache.spark.sql.DataFrame.我使用了将数据帧转换为rdd .rdd.处理完之后我想把它放回到数据帧中.我怎样才能做到这一点 ？
我想在spark中读取CSV并将其转换为DataFrame并将其存储在HDFS中 df.registerTempTable("table_name")
scala> val df = sqlContext.load("hdfs:///csv/file/dir/file.csv")
java.lang.RuntimeException: hdfs:///csv/file/dir/file.csv is not a Parquet file. expected magic number at tail [80, 65, 82, 49] but found [49, 59, 54, 10]
at parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:418)
at org.apache.spark.sql.parquet.ParquetRelation2$MetadataCache$$anonfun$refresh$6.apply(newParquet.scala:277)
at org.apache.spark.sql.parquet.ParquetRelation2$MetadataCache$$anonfun$refresh$6.apply(newParquet.scala:276)
at scala.collection.parallel.mutable.ParArray$Map.leaf(ParArray.scala:658)
at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply$mcV$sp(Tasks.scala:54)
at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:53)
at scala.collection.parallel.Task$$anonfun$tryLeaf$1.apply(Tasks.scala:53)
at scala.collection.parallel.Task$class.tryLeaf(Tasks.scala:56)
at scala.collection.parallel.mutable.ParArray$Map.tryLeaf(ParArray.scala:650)
at scala.collection.parallel.AdaptiveWorkStealingTasks$WrappedTask$class.compute(Tasks.scala:165)
at scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask.compute(Tasks.scala:514)
at scala.concurrent.forkjoin.RecursiveAction.exec(RecursiveAction.java:160)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
在Apache Spark中将CSV文件作为DataFrame加载的正确命令是什么？
RDD map和mapPartitions方法有什么区别？并且flatMap表现得像map或喜欢mapPartitions？谢谢.
(编辑)即,两者之间的差异(在语义上或在执行方面)
def map[A, B](rdd: RDD[A], fn: (A => B))
(implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
rdd.mapPartitions({ iter: Iterator[A] => for (i <- iter) yield fn(i) },
preservesPartitioning = true)
}
和:
def map[A, B](rdd: RDD[A], fn: (A => B))
(implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
rdd.map(fn)
}
root# bin/hadoop fs -mkdir t
mkdir: org.apache.hadoop.hdfs.server.namenode.SafeModeException: Cannot create directory /user/root/t. Name node is in safe mode.
无法在hdfs中创建任何内容
我做到了
root# bin/hadoop fs -safemode leave
但显示
safemode: Unknown command
问题是什么？
我知道https://wiki.apache.org/hadoop/AmazonS3的存在以及以下词语:
S3 Native FileSystem(URI scheme:s3n)用于在S3上读取和写入常规文件的本机文件系统.此文件系统的优点是您可以访问使用其他工具编写的S3上的文件.相反,其他工具可以访问使用Hadoop编写的文件.缺点是S3强加的文件大小限制为5GB.
S3A(URI方案:s3a)S3 Native的继承者s3n fs,S3a:系统使用亚马逊的库与S3交互.这允许S3a支持更大的文件(不超过5GB限制),更高性能的操作等等.文件系统旨在替代S3 Native的/后继者:只需替换URL模式,也可以从s3a访问从s3n:// URL访问的所有对象.
S3 Block FileSystem(URI scheme:s3)由S3支持的基于块的文件系统.文件存储为块,就像它们在HDFS中一样.这允许有效地实现重命名.此文件系统要求您为文件系统专用存储桶 - 不应使用包含文件的现有存储桶,也不应将其他文件写入同一存储桶.此文件系统存储的文件可能大于5GB,但它们不能与其他S3工具互操作.
为什么URI上的字母更改会产生这样的差异？例如
val data = sc.textFile("s3n://bucket-name/key")
至
val data = sc.textFile("s3a://bucket-name/key")
这种变化背后的技术差异是什么？有什么好文章我可以读到这个吗？
假设你已经在Hive中拥有"表",那么有没有像其他数据库一样快速获取该表的"CREATE"语句？