RDD管道每行一个外部进程
是指在分布式计算框架中，RDD（Resilient Distributed Datasets）管道操作可以将每个RDD的每一行作为一个独立的外部进程进行处理。
RDD是Spark中的核心数据结构，它代表了分布式的不可变数据集。RDD管道操作允许用户通过将每个RDD的每一行作为一个独立的外部进程来进行数据处理。这种方式可以提高计算效率，尤其适用于需要对每一行数据进行独立处理的场景。
使用RDD管道操作可以实现对大规模数据集的高效处理和转换。通过将每一行数据作为一个独立的外部进程处理，可以充分利用分布式计算框架的并行计算能力，提高数据处理的速度和效率。
RDD管道操作的优势包括：
- 高效处理：通过将每一行数据作为一个独立的外部进程处理，可以充分利用分布式计算框架的并行计算能力，提高数据处理的速度和效率。
- 灵活性：RDD管道操作可以与各种外部进程进行集成，可以根据实际需求选择合适的外部进程进行数据处理，提供了更大的灵活性和扩展性。
- 可靠性：RDD管道操作基于分布式计算框架，具有高可靠性和容错性，可以自动处理节点故障和数据丢失等问题。
RDD管道操作可以应用于各种场景，例如：
- 数据清洗和转换：可以使用外部进程对数据进行清洗和转换，例如过滤无效数据、格式转换等。
- 特征提取和处理：可以使用外部进程对数据进行特征提取和处理，例如提取文本特征、图像处理等。
- 数据分析和挖掘：可以使用外部进程对数据进行分析和挖掘，例如统计分析、机器学习等。
腾讯云提供了一系列与云计算相关的产品，以下是一些推荐的腾讯云产品和产品介绍链接地址：
- 云服务器（CVM）：提供弹性计算能力，支持多种操作系统和应用场景。详细介绍请参考：https://cloud.tencent.com/product/cvm
- 云数据库MySQL版（CDB）：提供高可用、可扩展的关系型数据库服务。详细介绍请参考：https://cloud.tencent.com/product/cdb
- 云原生容器服务（TKE）：提供高度可扩展的容器化应用管理平台。详细介绍请参考：https://cloud.tencent.com/product/tke
- 人工智能平台（AI Lab）：提供丰富的人工智能算法和工具，支持开发和部署智能应用。详细介绍请参考：https://cloud.tencent.com/product/ailab
- 物联网（IoT Hub）：提供物联网设备连接和管理的平台。详细介绍请参考：https://cloud.tencent.com/product/iothub
- 移动应用开发平台（MADP）：提供移动应用开发和管理的一站式解决方案。详细介绍请参考：https://cloud.tencent.com/product/madp
以上是关于RDD管道每行一个外部进程的完善且全面的答案，希望能对您有所帮助。
相关·内容
我正在开发一个火花应用程序，它需要从Kafka读取数据。我创造了一个卡夫卡主题，制片人在那里发布信息。我从控制台使用者那里验证消息已成功发布。
我写了一个简短的火花应用程序来读取Kafka的数据，但是它没有得到任何数据。以下是我使用的代码：
def main(args: Array[String]): Unit = {
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf().setAppName("SparkConsumer").setMaster(&
浏览 2提问于2016-11-28得票数 1
2回答
比较两种RDDs
、、、
我有两个RDD[ArrayString]，我们叫它们rdd1和rdd2。我将创建一个新的RDD，其中只包含rdd2的条目，而不是rdd1 (基于键)。我通过Intellij在Scala上使用星火。
我将rdd1和rdd2按一个键分组(我将比较这两个rdd1的键)：
val rdd1Grouped = rdd1.groupBy(line => line(0))
val rdd2Grouped = rdd2.groupBy(line => line(0))
然后，我使用了一个leftOuterJoin：
val output = rdd1Grouped.leftOuterJoin(rd
浏览 2提问于2016-06-22得票数 0
回答已采纳
1回答
从Spark输出数十亿行
、、、、
我正在尝试使用PySpark将大约5,000,000行的RDD输出为文本文件。这需要很长的时间，那么关于如何让.saveAsTextFile()更快有什么建议呢？
每行有3列，我将保存到HDFS。
浏览 0提问于2016-05-13得票数 0
我试图在Apache上运行一个外部c++脚本，使用rdd.pipe().I无法在文档中找到足够的信息，所以我在这里询问。
当使用rdd.pipe()时，外部脚本是否需要在集群中的所有节点上可用？
如果我没有在集群节点上安装任何东西的权限，该怎么办？是否有其他方法使脚本对工作节点可用？
浏览 1提问于2019-01-17得票数 2
回答已采纳
下面是我的代码片段。我有DStream，我试图将它保存到HDFS。只是想知道压缩的有效方法。
pairedDStream.foreachRDD { rdd =>
val time = Calendar.getInstance.getTimeInMillis;
val textOutputFolder = outputDir + "/output-" + time
if (args.length == 4) {
val compressionCodec = args(3)
rdd.saveAsTextF
浏览 2提问于2016-07-25得票数 0
我有一个有多个独立模块的pyspark程序，每个模块都可以独立地处理数据以满足我的各种需求。但它们也可以链接在一起，以便在管道中处理数据。这些模块中的每一个都构建了一个SparkSession，并各自完美地执行。
但是，当我试图在同一个python进程中连续运行它们时，我会遇到问题。在执行管道中的第二个模块时，second抱怨我试图使用的SparkContext已经停止：
py4j.protocol.Py4JJavaError: An error occurred while calling o149.parquet.
: java.lang.IllegalStateException: Ca
浏览 0提问于2017-01-05得票数 28
回答已采纳
我刚开始学习火花和斯卡拉。我在CDH 5.1.3上使用spark 1.0.0
我得到了一个名为dbTableKeyValueMap: RDD(String，String)的广播rdd，我想使用dbTableKeyValueMap来处理我的fileRDD(每行都有300+列)。这是代码：
val get = fileRDD.map({x =>
val tmp = dbTableKeyValueMap.lookup(x)
tmp
})
运行此本地挂起和/或在某个时候之后会出现错误：
scala.MatchError: null
at org.apache.spark.rdd.Pair
浏览 3提问于2016-01-07得票数 0
回答已采纳
1回答
我有以下结构的RDD：
my_rdd = [Row(text='Hello World. This is bad.'), Row(text='This is good.'), ...]
我可以使用python函数执行并行处理：
rdd2=my_rdd.map(lambda f: f.text.split())
for x in rdd2.collect():
print(x)
它给了我预期的输出。
然而，当我尝试使用火花-NLP语句破折器或情感分析器时，我会得到一个错误: PicklingError:未能序列化对象: TypeError:无法对_threa
浏览 7提问于2021-03-27得票数 0
def isSmallerScore(value:Int): Boolean ={
val const = 200
if(value < const) true else false
}
val rdd = sc.parallelize(Seq(("Java", 100), ("Python", 200), ("Scala", 300)))
val result1: RDD[(String, Int)] = rdd.filter(x => isSmallerScore(x._2))
val result2: RDD[(Stri
浏览 0提问于2021-02-23得票数 0
我正在用java编写一个包装程序，它应该通过在流中写入它们的标准并从它们的标准输出流中读取响应来将参数传递给其他进程。但是，当我试图传递的String太大时，PrintWriter.print只是阻塞。没有错误，只是结冰。有什么好的解决办法吗？
相关代码
public class Wrapper {
PrintWriter writer;
public Wrapper(String command){
start(command);
}
public void call(String args){
writer.pri
浏览 3提问于2015-07-24得票数 3
回答已采纳
我有一个pyspark.sql.dataframe，其中每行都是一篇新闻文章。然后，我有一个RDD，表示每篇文章中包含的单词。我想将单词的RDD作为名为' words‘的列添加到我的新文章数据框中。我试过了
df.withColumn('words', words_rdd )
但是我得到了一个错误
AssertionError: col should be Column
DataFrame如下所示
Articles
the cat and dog ran
we went to the park
today it will rain
但我有三千篇新闻文章。
我应用了一个清
浏览 0提问于2017-02-09得票数 5
1回答
我正在多次运行一个涉及spark并行化的程序。这个程序在最初的几次迭代中运行良好，但由于内存问题而崩溃。我使用的是Spark 2.2.0和Python2.7，我在亚马逊EC2上运行我的测试，内存为30g。
下面是我的Spark设置：
conf = pyspark.SparkConf()
conf.set("spark.executor.memory", '4g')
conf.set('spark.executor.cores', '16')
conf.set('spark.cores.max', '16&#
浏览 0提问于2017-10-12得票数 1
当我应用命令命令时，我没有得到任何输出。如何在使用命令或不同时使用两种情况时获得输出？
public class test
{
public static void main(String args[])
{
System.out.println(
systemtest("ifconfig | awk 'BEGIN { FS = \"\n\"; RS = \"\" } { print $1 $2 }' | sed -e 's/ .*inet addr:/,/' -e 's/ .*
浏览 2提问于2011-11-07得票数 0
回答已采纳
我是个新手，目前正在测试pySpark。它运行得比我预期的要慢，我想知道我是否正确地设置了它。
我的问题是：
我有一个由57个分区组成的RDD (每个分区~30 in )，所有分区都是缓存的(内存中的总大小为1700 in)。RDD包含13M字符串，每个字符串约300个字符。所以一般来说，不是一个大的数据集。那么为什么运行count()需要4秒呢？
我已经检查了UI，看来对于“计数”作业，它运行57个任务(正如预期的那样)，每个任务需要0.6秒，这对我来说非常慢。
我在Google云上运行，在Mesos之上，有1名主人和2名奴隶。每个实例都有8个核和30 GB的RAM。
我的问题：
每项任
浏览 0提问于2015-04-02得票数 1
我如何使外部工作更像一个本地Jenkins作业，特别是包含进度'bar‘的Jenkins仪表板？
我遇到过，但这似乎只提供了一种说法
我的作业已经用这个控制台输出和这个结果代码完成了。
我也希望看到目前的进展。换句话说，我希望我的外部工作能够以类似于本地Jenkins工作的方式出现在类似的工作上。
那么，是否有可能推出比外部监控工作建议的更多的数据呢？
浏览 0提问于2018-11-02得票数 0
我试图将数据从RDD[string]中提取到另一个RDD[string]中
RDD包含与此类似的数据：
17.808 15.749 6.649 -0.548 15.9994
我需要将每行的第4和第5字段相乘，并将它们存储到不同的RDD[string]中。
我可以使用下面的代码提取一个字段
ansRDD = rawRDD(._split(" ")(4)).(_.toFloat)
rawRDD包含字符串。
但我需要将这两个字段提取到一个RDD中，作为
-0.548 15.9994
这样我就可以简单地
answer = ansRDD.foreach(case(a,b) =
浏览 0提问于2015-12-03得票数 1
回答已采纳
2回答
我试图以下面描述的方式(使用PySpark)在地图中使用查找(在spark中)，但得到一个错误。
这是Spark不允许做的事情吗？
>>> rdd1 = sc.parallelize([(1,'a'),(2,'b'),(3,'c'),(4,'d')]).sortByKey()
>>> rdd2 = sc.parallelize([2,4])
>>> rdd = rdd2.map(lambda x: (x, rdd1.lookup(x)))
>>> rdd.co
浏览 1提问于2016-03-28得票数 0
在s3中，我有许多格式为log-.tar.gz的文件。我想要处理它们，处理它们(从每行提取一个字段)，并将其存储在一个新文件中。
我们有很多方法可以做到这一点。一种简单方便的方法是使用textFile方法访问文件。
//Read file from s3
rdd = sc.textFile("s3://bucket/project_name/date_folder/logfile1.*.gz")
我关心的是集群的内存限制。这样，主节点将会过载。对于可以按集群类型处理的文件大小，是否有任何粗略的估计？
我想知道是否有一种方法可以将从s3获取*.gz文件的过程并行化，因为它们已经按
浏览 0提问于2015-11-13得票数 0
不运行过滤器(如验证)的理由是什么：
ServiceStackController.Execute(requestDto)
从MVC控制器？
我可以理解，如果您从MVC控制器手动解析一个服务并调用一个方法，过滤器将不会被执行。
using (var hello = HostContext.
ResolveService<HelloService>(base.HttpContext))
{
hello.Get(name);
}
但是希望过滤器/验证能够激发Execute方法.
浏览 0提问于2016-04-06得票数 1
回答已采纳
我使用的是PySpark (Python3.5.2和Spark2.2.0.2.6.4.0-91)，我有一个预测值的数据帧(通过用MLlib库定义的随机森林模型)，结构如下： DataFrame[id: bigint, features: vector, rawPrediction: vector, probability: vector, prediction: double] 我是这样得到的： rf_predictions = random_forest_model.transform(dataframe) 但是当我想要显示它的内容时，它只适用于前两列"id“和”feature“：
浏览 34提问于2019-09-21得票数 0
2回答
我有一个rdd，格式为每行(key, (int, double))
我想将rdd转换为(key, ((int, double), (int, double) ...) )。
其中，新rdd中的值是按双精度排序的前N个值对
到目前为止，我想出了下面的解决方案，但它真的很慢，并且永远运行，它在较小的rdd上工作得很好，但现在rdd太大了
val top_rated = test_rated.partitionBy(new HashPartitioner(4)).sortBy(_._2._2).groupByKey()
.mapValues(x => x.takeRig
浏览 0提问于2018-10-14得票数 1
我正在使用Erlang端口读取Linux进程的输出。每当我连接的Erlang进程终止时，我希望Linux进程自动终止。从文档来看，在我看来，这应该自动发生，但事实并非如此。
最小示例。将其放入文件test.erl中：
-module(test).
-export([start/0, spawn/0]).
start() ->
Pid = spawn_link(?MODULE, spawn, []),
register(test, Pid).
spawn() ->
Port = open_port({spawn, "watch date"}
浏览 1提问于2021-05-26得票数 2
1回答
我有一个问题需要澄清，还有两个使用云数据融合的错误:背景:创建一个管道，使用云数据融合将数据从Oracle中的单个表(本地服务器11.2.0.4版本)移动到BigQuery中。我从即时客户端11.2.0.4.0下载了Oracle JDBC驱动程序，并将ojdbc6.jar文件用作下面部署的oracle驱动程序。如果我使用一个更新的，我会得到一个日期错误。
清晰度：似乎有3个地方可以加载OJDBC驱动程序：
(a) Wrangler数据库列表(例如: sbl)，单击编辑，它将要求我安装Oracle驱动程序或Oracle瘦驱动程序。我需要输入Oracle (我复制了示例oracle.jdbc.dr
浏览 4提问于2020-01-21得票数 2
1回答
当我运行以下脚本时，它完全按照我的要求执行并退出：
setDisplay.sh：
#!/bin/bash
Xvfb -fp /usr/share/fonts/X11/misc/ :22 -screen 0 1024x768x16 2>&1 &
export DISPLAY=:22
当我运行./setDisplay.sh时，一切都运行正常。
好了，这就是有趣的开始。
我有一个调用setDisplay的Perl脚本...
下面是eamorr.pl脚本：
#!/usr/bin/perl
use strict;
use warnings;
my $homeDir="
浏览 1提问于2013-06-21得票数 1
回答已采纳
我正在尝试为文本分类训练一个ANN：
mlp = MultilayerPerceptronClassifier(maxIter=10, layers=[5,3], blockSize=128, seed=123)
model_stacking = mlp.fit(input_vector.select(['features', 'label']))
preditions_foo = model_stacking.transform(validation)
predition = evaluator.evaluate(preditions_foo)
当应用fit(
浏览 2提问于2019-06-27得票数 1
3回答
我能把两个进程相互输送到一起吗？
、、、、
在此页中，从4.4BSD操作系统的设计和实现来看，可以说：
管道和套接字之间的一个主要区别是，管道需要一个公共父进程来设置通信通道。
但是，如果我记录正确，那么创建一个新流程的唯一方法就是对一个现有流程进行fork。所以我真的不明白为什么两个进程不能有一个共同的祖先。那么，我认为任何一对过程都可以通过管道传递给对方，这是正确的吗？
浏览 0提问于2013-05-12得票数 8
回答已采纳
1回答
我在类似的用例中使用了transform方法，如Transform 节中所描述的
spamInfoRDD = sc.pickleFile(...) # RDD containing spam information
# join data stream with spam information to do data cleaning
cleanedDStream = wordCounts.transform(lambda rdd: rdd.join(spamInfoRDD).filter(...))
我的代码如下：
sc = SparkContext("local[4]",
浏览 2提问于2015-06-17得票数 4
1回答
多处理RDDs列表
、、、
我正在尝试多进程，如下所示
from pyspark.context import SparkContext
from multiprocessing import Pool
def square(rdd_list):
def _square(i):
return i*i
return rdd_list.map(_square)
sc = SparkContext('local', 'Data_Split')
data = sc.parallelize([1,2,3,4,5,6])
dataCollection = [d
浏览 0提问于2016-07-14得票数 5
我在HDFS上有一个文件，格式为：
61,139,75
63,140,77
64,129,82
68,128,56
71,140,47
73,141,38
75,128,59
64,129,61
64,129,80
64,129,99
我用它创建了一个RDD，并用它们的索引压缩了元素：
val data = sc.textFile("hdfs://localhost:54310/usrp/sample.txt")
val points = data.map(s => Vectors.dense(s.split(',').map(_.toDouble)))
浏览 8提问于2016-09-17得票数 2
回答已采纳
有什么方法可以访问由FUSE文件系统进程创建的数据吗？例如，在prefix_write()中，我将一些数据存储在内存中，并希望从另一个进程访问这些数据。
共享内存应该可以工作。但我正在寻找一种更优雅的解决方案，比如fuse_operations中的自定义字段，我可以从其他进程中将其作为函数访问。但据我所知，fuse_operations中的字段必须来自POSIX，因此可能无法做到这一点。如果我说错了，请纠正我。
谢谢
浏览 0提问于2011-07-18得票数 0
1回答
我有一个文本文件，每行有两个用空格分隔的整数值
假设我有两个RDDs。一个具有一对值(x，y)，其中x和y是2个整数值。另一个RDD (rdd2 -> (a，b))，它只有一个具有两个整数值a，b的对。
rdd1 = textfile.map(lambda line: line.split())
...
我的目标是用rdd2中的相应值减去rdd1中的每个值
结果将如下所示
x1-a,y1-b
x2-a,y2-b
x3-a,y3-b
...
我怎样才能做到这一点？
浏览 0提问于2015-10-22得票数 0
1回答
dataframe中的列"colA“包含整数值：
+-----+
| colA|
+-----+
| 1|
| 2|
| 1|
| 3|
+-----+
这些整数值可以通过Redis字典映射到字符串：
+----+------+
| key| value|
+----+------+
| 1| a|
| 2| b|
| 3| c|
+----+------+
我需要创建一个新列"colB“，它将包含"colA”到字符串值的映射：
+-----+-----+
| colA| colB|
+-----+-----+
浏览 0提问于2018-02-05得票数 1
回答已采纳
1回答
管道、叉和非阻塞IPC
、、、
因此，我试图在一个子进程中运行一个系统命令(或者exec，或者其他什么)，然后将一些输入推送到它，然后得到它的输出。它看上去就像叉子()之后的样子，pc和cp是父-子管道和子-父管道。
case 0:
/* Child. */
close(STDOUT_FILENO); /* Close current stdout. */
dup2(cp[1], STDOUT_FILENO);
close(STDIN_FILENO);
dup2(pc[0], STDIN_FILENO);
close( p
浏览 1提问于2011-07-08得票数 3
我需要打电话从我的EMR火花工作外部进程。我看到rdd.pipe将允许我将一个RDD输送到一个进程。(顺便说一句，这是每个RDD的一个进程，还是每个元素的一个？)。
但是，我的外部进程需要一个文件名作为输入，并生成一个文件作为输出。
如何调用此外部进程并随后将输出文件作为RDD加载？
浏览 3提问于2016-09-15得票数 0
回答已采纳
3回答
我的spark应用程序使用的是numpy数组的RDD。
目前，我正在从亚马逊网络服务S3读取我的数据，并将其表示为一个简单的文本文件，其中每行都是一个向量，每个元素由空格分隔，例如：
1 2 3
5.1 3.6 2.1
3 0.24 1.333
我使用numpy的函数loadtxt()来创建一个numpy数组。
然而，这种方法似乎非常慢，并且我的应用程序花费了太多的时间(我认为)来将我的数据集转换为numpy数组。
你能给我推荐一种更好的方法吗？例如，我应该将我的数据集保存为二进制文件吗?我应该以另一种方式创建RDD吗？
我如何创建我的RDD的一些代码：
data = sc.textFile(
浏览 0提问于2015-11-19得票数 3
回答已采纳
2回答
我有一个HadoopRDD，我用简单的Map函数创建第一个RDD，然后用另一个简单的Map函数从第一个RDD创建第二个RDD。类似于：
HadoopRDD -> RDD1 -> RDD2.
我的问题是，Spak是否会逐条记录地遍历HadoopRDD记录以生成RDD1，然后它将逐条记录地遍历RDD1记录以生成RDD2，还是在HadoopRDD上进行迭代，然后生成RDD1，然后一举生成RDD2。
浏览 4提问于2015-03-06得票数 1
回答已采纳
1回答
我正在努力编写一个c++代码，该代码将一些“字符串”传递给外部程序并获得反馈。
例如，我想将一个c++代码传递给gcc，并将错误消息返回到我的原始代码。
如何用c++编写
浏览 1提问于2013-09-30得票数 1
2回答
从Java运行PHP
、、、
我正在尝试从java文件中运行以下代码。
我正在尝试显示php版本，稍后将更改为运行php文件。
Process p = Runtime.getRuntime().exec("cmd /C PHP/php.exe -v");
BufferedReader in = new BufferedReader(new InputStreamReader(
p.getInputStream()));
String line = null;
while ((line = in.readLine()) != null) {
浏览 2提问于2012-11-29得票数 0
1回答
我正试图将Python作业提交给一个2个工作节点星火库集群，但始终看到以下问题，它最终会导致火花提交失败：
15/07/04 21:30:40 WARN scheduler.TaskSetManager: Lost task 0.1 in stage 0.0 (TID
2, workernode0.rhom-spark.b9.internal.cloudapp.net):
org.apache.spark.SparkException: Python worker did not connect back in time
at org.apache.spark.api.pyt
浏览 0提问于2015-07-04得票数 1
回答已采纳
我在windows机器中使用gzip实用程序。我压缩了一个文件并以blob的形式存储在DB中。当我想使用gzip实用程序解压缩这个文件时，我将这个字节流写入process.getOutputStream。但是在30 to之后，它无法读取该文件。挂在那里。
尝试使用内存参数，读取和刷新逻辑。但是，同样的数据，如果我试图写入一个文件，它是相当快的。
OutputStream stdin = proc.getOutputStream();
Blob blob = Hibernate.createBlob(inputFileReader);
InputStream source = blob.g
浏览 2提问于2010-11-13得票数 3
1回答
我正在尝试写入stdin，并从外部程序读取stdout (和stderr )，而不更改代码。
我尝试过使用命名管道，但stdout直到程序终止才会显示，并且stdin只在第一个输入上起作用(然后cin为空)。
我尝试过使用/proc/pid/fd，但它只从终端而不是程序中写入和读取。
我试着为此编写了一个字符设备文件，它工作了，但一次只有一个程序(这需要在同一时间为多个程序工作)。
在这一点上，据我所知，我可以编写在多个程序中多路复用io的驱动程序，但我不认为这是“正确”的解决方案。
其主要目的是通过web界面查看节目的馈送。我相信一定有什么方法可以做到这一点。有没有什么我以前没试过的东西？
浏览 1提问于2013-03-20得票数 2
回答已采纳
1回答
我想知道的是，我可以从Java读取外部进程的输出流吗？假设这个外部进程是一个windows终端，并且不是使用ProcessBuilder或Runtime.exec()从Java启动的。
浏览 1提问于2013-09-19得票数 2
1回答
火花管实例
、、、
我对Spark还不熟悉，我想弄清楚管道方法是如何工作的。我在Scala中有以下代码
sc.textFile(hdfsLocation).pipe("preprocess.py").saveAsTextFile(hdfsPreprocessedLocation)
hdfsLocation和hdfsPreprocessedLocation值都很好。作为证明，下面的代码可以从命令行工作
hadoop fs -cat hdfsLocation/* | ./preprocess.py | head
当我运行上面的火花代码时，我会得到以下错误
14/11/25 09:41:50 ERROR
浏览 2提问于2014-11-25得票数 3
回答已采纳
我正在通过Google Data Fusion Pipeline处理文件，但随着管道的进行，我得到了以下警告和错误：
09/25/2020 12:31:31警告任务6的org.apache.spark.storage.memory.MemoryStore#66-Executor任务启动工作进程没有足够的空间在内存中缓存rdd_10_6！(到目前为止计算了238.5 MB )
09/25/2020 12:45:05错误org.apache.spark.scheduler.cluster.YarnClusterScheduler#70-dispatcher-event-loop-1
主机cdap-
浏览 2提问于2020-09-25得票数 0
1回答
APUE的下图让我不禁要问:在流程会话中，会话领导人的流程组是否只包含会话领导者，而不包含其他流程？
fork()创建的任何进程都将从其父进程继承相同的进程组和会话。那么，会话领导者可以将任意数量的进程分成自己的进程组吗？
例如，shell (当它是会话领导者时)是它的进程组中的唯一进程吗？如果是，是因为当shell分叉子进程时，子进程最初与shell位于同一个进程组中，但随后是立即通过调用setpgid()？
谢谢。
📷
浏览 0提问于2018-05-05得票数 4
回答已采纳
1回答
我正在尝试使用spark (python3.5)来并行化一个matlab程序，但是我遇到了一些问题。我不知道这是不是一个互操作性问题，因为对象的类型可能在matlab和python之间不匹配，或者如果我有另一个实习生问题。 from pyspark import SparkConf, SparkContext
import matlab.engine
import numpy as np
if __name__ == "__main__":
conf = SparkConf().setAppName("reduce").setMaster("
浏览 76提问于2019-04-28得票数 1
回答已采纳
在我的sparkJob中，我执行了几个转换和聚合操作。我需要在管道操作之前和之后通知服务。具体地说，参考下面的代码，我需要在RDD中的所有文档上完成“管道操作1”之后，在“管道操作2”开始之前进行一个服务调用。
JavaRDD<Document> stage_y = stage_x.
map(r -> someRddOperation(r) ).
pipe( "pipe operation 1 to external process" ).
map(r ->
浏览 0提问于2018-09-27得票数 0
1回答
我想在pyspark应用程序中使用预先训练好的嵌入模型(fasttext)。 因此，如果我广播该文件(.bin)，则会抛出以下异常: Traceback (最近一次调用)： cPickle.PicklingError: Could not serialize broadcast: OverflowError: cannot serialize a string larger than 2 GiB 相反，我尝试使用sc.addFile(modelpath) where modelpath=path/to/model.bin，如下所示： 我创建了一个名为fasttextSpark.py的文件 i
浏览 36提问于2019-01-28得票数 5
我有一个安装了PySpark的管道，我正将其保存到磁盘上以备将来使用。
下面是我的流水线代码：
model = Pipeline(stages=[segment_indexer, model_name_indexer, make_name_indexer, engine_type_indexer, segment_encoder, model_name_incoder, make_name_incoder, engine_type_incoder, x_assembler, estimator]).fit(trainingData)
model.save('file:/opt/app
浏览 31提问于2018-02-22得票数 3
回答已采纳
我有一个80 rdd的列表，我想要处理，然后最终加入。"process“部分包括为每个rdd执行一个映射和一个约简键。那我就通过联合加入他们。下面是我的代码草图：
rdds0.foreach(_.persist()) //rdds0 are persisted
//trigger a map and a shuffle for each rdd
vals rdds = rdds0.map(rdd => rdd.map(f1).reduceByKey(f2))
//action on the union of the rdds
sparkContext.union(rdds).
浏览 1提问于2018-07-19得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
