在Spark流作业内部调用实用程序(外部)
在Spark流作业内部调用实用程序(外部)是指在Spark流作业中调用外部实用程序来完成特定的任务或功能。这种方式可以通过调用外部程序来扩展Spark流作业的功能，提高作业的灵活性和可扩展性。
优势：
- 扩展功能：通过调用外部实用程序，可以方便地扩展Spark流作业的功能，满足特定需求。
- 灵活性：使用外部实用程序可以根据具体需求选择最适合的工具或库，灵活地完成任务。
- 可复用性：外部实用程序可以独立于Spark流作业进行开发和维护，可以在多个作业中复用，提高开发效率。
应用场景：
- 数据清洗和预处理：可以使用外部实用程序来清洗和预处理数据，例如使用Python的Pandas库进行数据清洗和转换。
- 机器学习和数据挖掘：可以调用外部实用程序来执行机器学习算法或数据挖掘任务，例如使用Scikit-learn库进行特征提取和模型训练。
- 图像和音视频处理：可以使用外部实用程序来处理图像和音视频数据，例如使用OpenCV库进行图像处理或FFmpeg库进行音视频编解码。
- 自定义数据处理：可以根据具体需求开发自定义的外部实用程序，例如使用Java或C++编写高性能的数据处理程序。
推荐的腾讯云相关产品：
腾讯云提供了多个与Spark流作业相关的产品和服务，可以帮助用户更好地进行实用程序的调用和管理。
- 腾讯云Spark：腾讯云提供了托管的Spark集群服务，用户可以在集群中运行Spark流作业，并通过调用外部实用程序来完成特定任务。 产品链接：https://cloud.tencent.com/product/spark
- 腾讯云函数计算：腾讯云函数计算是一种事件驱动的无服务器计算服务，用户可以在函数中调用外部实用程序来处理事件触发的任务。 产品链接：https://cloud.tencent.com/product/scf
- 腾讯云容器服务：腾讯云容器服务提供了容器化的应用部署和管理服务，用户可以在容器中运行Spark流作业，并调用外部实用程序进行扩展。 产品链接：https://cloud.tencent.com/product/tke
请注意，以上推荐的产品仅为示例，具体选择应根据实际需求和情况进行。
相关·内容
1回答
我在我的火花流应用程序中使用createDirectStream。我将批处理间隔设置为7秒，大部分时间批处理作业可以在大约5秒内完成。但是，在非常罕见的情况下，批处理作业需要花费60秒，这将延迟一些批作业。为了减少总延迟时间，我希望我能处理更多的流数据，这些数据分布在一次延迟的工作上。这将有助于流尽快恢复正常。
因此，我想知道有什么方法可以在延迟出现时动态地更新/合并spark和kafka输入的批次大小。
浏览 3提问于2016-12-19得票数 0
我正在尝试执行一个以Twitter为源的Spark Streaming示例，如下所示：
public static void main (String.. args) {
SparkConf conf = new SparkConf().setAppName("Spark_Streaming_Twitter").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaStreamingContext jssc = n
浏览 0提问于2014-07-02得票数 26
回答已采纳
我正在创建流分析应用程序使用火花，Flink和卡夫卡。每个分析/功能都将实现为Microservice，以便该分析可以在以后的不同项目中使用。
我在简单的Scala应用程序中完美地运行了我的Spark/Flink作业，并分别在Spark & Flink集群上提交了这个作业。但是，当REST将startJob()请求调用到web服务时，我必须启动/运行此作业。
如何在面向web服务的应用程序中集成我的Spark & Flink数据处理功能？
直到现在，我尝试了，但我发现了很多问题，您可以检查
我认为我对流处理微服务应用没有采取正确的方向。寻找正确的方向来实现这
浏览 3提问于2017-04-25得票数 1
我有多个谷歌数据流作业，用于数据收集和ETL目的。然后google dataproc作业(Spark)用于进一步的机器学习。
我想把这些工作像工作流一样绑在一起，然后我就可以安排整个工作流程了。
你有什么建议/产品可以帮助我吗？
浏览 0提问于2016-05-26得票数 2
我正在使用Spark 3.0.1，并在Kubernetes上进行项目spark部署，其中Kubernetes代理spark作业的集群管理器，spark使用客户端模式提交作业。如果群集没有足够的资源(CPU/内存)用于最小数量的执行器，则执行器将无限期地处于挂起状态，直到资源释放。 假设集群配置为： total Memory=204Gi
used Memory=200Gi
free memory= 4Gi
SPARK.EXECUTOR.MEMORY=10G
SPARK.DYNAMICALLOCTION.MINEXECUTORS=4
SPARK.DYNAMICALLOCATION.MAXEXEC
浏览 7提问于2021-02-05得票数 0
我的火花结构化流作业不断生成拼花文件，我想在过期后删除(假设在30天后)。
我存储分区键为cron 3339/I 8601中事件日期的分区数据，以便在HDFS级别上基于cron作业(根据字符串比较删除所有带有分区键的文件夹< oldestAllowedAge )轻松地进行内务管理。
但是，由于我引入了Spark流，Spark将元数据写入到拟写入数据本身旁边的名为_spark_metadata的文件夹中。如果我现在只是删除过期的HDFS文件，并在整个数据集上运行一个火花批处理作业，由于找不到文件，作业将失败。批处理作业将读取元数据，并期望已经删除的文件存在。
解决这个问题的简单方法就是禁用
浏览 6提问于2019-04-11得票数 4
回答已采纳
1回答
我正在将现有的管道从spark 1.6.0迁移到spark 2.1.0(cdh 5.15.1)。我使用的oozie版本(4.1.0)不支持spark2操作，所以我们使用java操作来运行spark2作业。
作业可以通过java action成功执行，但我面临的一个问题是，每当oozie工作流被终止时，spark应用程序都不会被终止，尤其是在运行集群模式时。
我可以理解java操作在单独的容器中启动spark驱动程序，这是一个单独的jvm进程。我只是想知道是否有办法处理这种情况。
浏览 1提问于2019-03-30得票数 0
1回答
我正在使用微批处理(readStream)从Kafka流中读取消息，并通过writeStream处理它们并将结果写入另一个Kafka主题。作业(流查询)设计为“永远”，处理10秒大小的微批(处理时间)。设置了checkpointDirectory选项，因为Spark需要检查点。
但是，当我尝试提交具有相同源流(相同主题等)的另一个查询时，但是可能有不同的处理算法)，Spark完成了前面正在运行的查询，并创建了一个具有相同ID的新查询(因此它从前一个作业“完成”的完全相同的偏移量开始)。
如何告诉Spark，第二个作业与第一个作业不同，因此不需要从检查点恢复(即，意图是创建一个与前一个任务没有连
浏览 4提问于2021-03-05得票数 0
回答已采纳
有很多使用Kubernetes Spark操作符创建Spark作业并简单地提交具有以下内容的请求的示例
kubectl apply -f spark-pi.yaml
可以在this 中找到spark-pi.yaml
有没有人知道用Java提交这样的作业最简单的方法是什么？
浏览 15提问于2020-06-04得票数 0
在条件语句中结束spark作业的推荐方式是什么？
我正在对我的数据进行验证，如果为false，我希望优雅地结束spark作业。
现在我有：
if (isValid(data)) {
sparkSession.sparkContext.stop()
}
但是，我得到以下错误：
Exception in thread "main" java.lang.IllegalStateException: SparkContext has been shutdown
然后它会显示一个堆栈跟踪。
sparkContext.stop()难道不是优雅地结束spark作业的正确方式吗？
浏览 56提问于2017-12-09得票数 0
我们正在试图找到一种加载Spark (2.x) ML训练模型的方法，这样我们就可以根据请求(通过REST接口)查询它并获得预测，例如
有一些现成的库可以将模型加载到Spark中(假设它是在使用MLWritable训练后存储在某个地方)，然后将其用于预测，但由于SparkContext的初始化，将其包装在作业中并在每次请求/调用时运行它似乎有些过分。
然而，使用Spark的好处是我们可以保存我们的管道模型并执行相同的功能转换，而不必在SparkContext外部实现它。
经过一些挖掘，我们发现可以潜在地帮助我们解决这个问题，它允许我们为作业服务器初始化一个“热”的spark-context，因此
浏览 1提问于2017-02-14得票数 5
2回答
有没有办法在Spark Structured Streaming中执行outer_join (特别是左外部连接)？我注意到了这个，它支持对流数据集进行内连接(它工作得很好！)。我想知道outer_join是否也有类似的东西。我尝试使用JIRA中的相同逻辑进行外部连接，但无法获得正确的输出。谢谢!
浏览 2提问于2018-01-26得票数 1
回答已采纳
为什么rundeck不启动调度的spark作业，即使之前的作业仍在执行？ Rundeck正在跳过设置为在执行前一个作业期间启动的作业，然后在其执行完成后根据计划启动新作业。 但我希望启动一个计划的作业，即使前一个作业正在执行。
浏览 20提问于2019-06-21得票数 1
我有一个奇怪的问题，试图读取数据从卡夫卡使用火花结构化流。我的用例是能够从最大的/最新的偏移量中读取主题。
我的阅读吐露：
val data = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "some xyz server")
.option("subscribe", "sampletopic")
.option("auto.offset.reset",
浏览 1提问于2018-02-14得票数 4
我正在编写一个Spark (v2.2)批处理作业，它从Kafka主题中读取。Spark作业正在使用cron进行调度。我不能使用Spark Structured Streaming，因为不支持非基于时间窗口。
val df = spark
.read
.format("kafka")
.option("kafka.bootstrap.servers", "...")
.option("subscribe", s"kafka_topic")
我需要设置ka
浏览 15提问于2017-08-28得票数 2
回答已采纳
我有一个场景--每次使用同名的实用程序数据文件--先删除旧文件，然后加载更新的/最新的文件，然后将更新的/最新的文件从源系统中定期推送到HDFS中。该数据文件需要进一步推入一个单元表中。
我正在寻找实现这一目标的最佳途径：
1)在单元格中创建一个外部表，指向HDFS中的数据由HDFS中的实用程序推送数据的位置，这将在数据文件被该实用程序更新后立即自动更新该数据。
2)使用蜂窝仓库中的数据创建一个内部/托管表。现在，使用配置为cron作业的shell脚本--在正常间隔内，检查显示在单元仓库中的文件的时间戳，以及位于实用程序推送/更新数据的位置的文件的时间戳，如果时间戳不一样，则脚本将从蜂窝仓库
浏览 7提问于2017-11-19得票数 0
spark作业中有多个查询，当我开始这个作业时，我发现每个查询都会与kafka建立连接，并且它们之间不共享数据，所以如何缓存数据以避免多次读取相同的数据。 我曾尝试使用命令“cache table cache_table;”缓存表，然后 Queries with streaming sources must be executed with writeStream.start();;
kafka
org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.org$apache$spark$sql$catalyst$a
浏览 34提问于2019-01-30得票数 0
我想使用Dataproc在Google Cloud上创建一个摄取/聚合流程，在此之前，我每天/小时都希望在收集的数据上运行Spark作业。 有没有办法安排Spark作业？或者基于例如到达流上的任何新的数据事件来制作该触发器？
浏览 23提问于2020-08-24得票数 2
回答已采纳
我编写了Kafka流应用程序，我想将它部署到Kafka集群上。因此，我构建了一个jar文件并使用以下命令运行它：
java -jar KafkaProcessing-1.0-SNAPSHOT-jar-with-dependencies.jar testTopic kafka1:9092,kafka2:9092 zookeeper1:2181,zookeeper2:2181 output
它运行正常，但作业运行在我在命令上运行的机器上！我想当我指定BOOTSTRAP-SERVERS时，它会自动在集群上进行计算，而不是在主机上！
那么，我的问题是如何在Kafka集群上提交kafka流作业呢？就像
浏览 1提问于2017-12-01得票数 5
回答已采纳
一般来说，如何在不输出文件的情况下从spark作业中提取结果？
比方说，我有一个scala程序，它创建一个spark launcher作业。
我可以用Futures完成这个spark任务。一旦spark作业完成，我如何才能得到结果呢？(我之前的一种方法是让spark作业将结果写到一个文件中，然后外部scala程序读取该文件。<-我想避免这种策略)
浏览 1提问于2015-12-23得票数 0
我有一个AWS S3文件夹，其中存储了大量的JSON文件。我需要ETL这些文件与AWS EMR通过星火和存储转换为AWS RDS。
为此，我在Scala上实现了Spark作业，一切都很好。我计划每周执行一次这项工作。
外部逻辑有时可以向AWS S3文件夹中添加一个新文件，因此下次当我的火花作业开始时，我只想处理新的(未处理的) JSON文件。
现在，我不知道在哪里存储有关已处理的JSON文件的信息，以便Spark作业可以决定要处理哪些文件/文件夹。请您告诉我，使用Spark/AWS跟踪这种变化的最佳实践是什么(以及如何)？
浏览 0提问于2018-10-15得票数 1
回答已采纳
我正在使用Spark 3.0.2，我有一个流作业，它消耗Kafka中的数据，触发持续时间为"1分钟“。
我在Spark UI中看到，按照定义，每1分钟就会有一个新作业，但我看到onQueryProgress方法每5到6分钟被调用一次。我认为这个方法应该在每个微批之后直接调用。
有没有办法控制这个持续时间，使其等于触发持续时间？
浏览 8提问于2021-04-19得票数 1
1回答
星火结构流蓝/绿部署
、、、、
我们希望能够部署我们的星火作业，这样在部署期间处理数据就不会有任何停机时间(目前大约有2-3分钟的窗口)。在我看来，最简单的方法就是模拟“蓝色/绿色部署”理念，那就是旋转星火作业的新版本，让它热身，然后关闭旧作业。但是，使用结构化流和检查点，我们不能这样做，因为新的Spark作业看到最新的检查点文件已经存在(从旧作业)。我在下面附上了一个样本错误。有没有人对潜在的解决办法有任何想法？
我考虑过将现有的检查点目录复制到新创建的作业的另一个检查点目录中--尽管这应该是一种解决办法(一些数据可能会被重新处理，但我们的DB应该去复制)，但这似乎非常麻烦，而且是我不想追求的。
Caused by: or
浏览 0提问于2018-04-04得票数 5
1回答
我有一个星火流作业运行在一个集群(Spark1.6)上，它是S3的检查点。当我刚开始工作时，我可以看到“流”选项卡。但是，当我从检查点重新启动作业时，流选项卡就会消失。作业仍然作为流作业工作，我看到批处理出现在已配置的批处理间隔中。见下文。
如果清除检查点数据，选项卡将返回。我怀疑流选项卡在从检查点重新启动时没有正确注册。
我看了看。当应用程序状态从检查点反序列化时，是否可能不调用此流？
有人知道怎么解决这个问题吗？
浏览 4提问于2016-04-18得票数 2
回答已采纳
这是我在yarn集群环境中执行的spark程序。
// Making a call to java program from spark scala program
val affiliationRowRDD = affiliationDistinctRowRDD.map(a =>Row.fromSeq(
datavolveFromJavaJob.callDatavolveFromJavaJob(a.toString()).split(",").toSeq
))
//printing count and records of rdd - This
浏览 13提问于2015-12-20得票数 0
我正在尝试运行以下简单的Spark代码：
public static void main(final String[] args)throws Exception {
ClassLoader.getSystemClassLoader().getClass().getName();
Thread.currentThread().getContextClassLoader().getClass().getName();
GraphApiApplication.class.getClassLoader();
/*if (args.length < 1) {
浏览 7提问于2017-07-05得票数 0
1回答
我正尝试在Amazon AWS中运行一个简单的pyspark作业，它被配置为通过spark-default.conf文件使用Yarn。我对Yarn部署代码有点困惑。 我看到一些示例代码如下所示： conf = SparkConf()
conf.setMaster('yarn-client')
conf.setAppName('spark-yarn')
sc = SparkContext(conf=conf) 在这种情况下，当指定了'yarn-client‘时，我不确定我应该如何执行spark作业。我通常这样做： $spark-submit --depl
浏览 3提问于2020-01-27得票数 1
回答已采纳
3回答
在Spark2.2.0中，API中有新的端点用于获取关于流作业的信息。
我在EMR集群上运行Spark，在集群模式下使用Spark2.2.0。
当我命中流作业的端点时，它给我的只是错误消息：
没有连接到的流侦听器
我已经研究了一下星火代码库，但是这个特性并没有很好的文档化。所以我很好奇这是不是个虫子？为了使这个端点正常工作，我需要做一些配置吗？
在集群上运行时，这似乎是一个具体的问题。运行在我的本地机器上的Spark2.2.0上的相同代码显示了预期的统计信息，但在集群上运行时给出了错误消息。
浏览 1提问于2017-12-12得票数 2
回答已采纳
火花作业的定义是：
作业--由多个任务组成的并行计算，这些任务响应于火花操作(例如，保存、收集)；您将在驱动程序的日志中看到这个术语。
那么，为什么每个spark-submit在dataproc控制台中只创建一个作业id？
示例:下面的应用程序应该有两个火花作业
sc.parallelize(range(1000),10).collect()
sc.parallelize(range(1000),10).collect()
浏览 3提问于2022-01-21得票数 2
回答已采纳
2回答
我有一个Dataproc集群：
master -6核心
工作人员{0-7}-6核心
最大分配:内存:24576，vCores:6
有两个火花流作业要提交，一个接一个
首先，我尝试使用默认配置提交spark.dynamicAllocation.enabled=true。
在30%的情况下，我发现第一个作业几乎捕获了所有可用内存，第二个任务排队等待了很长时间的资源。(这是一个流作业，每批占用一小部分资源)。
我的第二次尝试是改变动态分配。我提交了相同的两个配置相同的作业：
spark.dynamicAllocation.enabled=false
spark.executor.memory=12g
浏览 2提问于2019-03-04得票数 1
回答已采纳
1回答
火花流的动态分配
、、、、
我有一个星火流作业运行在我们的集群与其他作业(星火核心作业)。我想使用动态资源分配这些工作，包括星火流。根据下面的JIRA问题，动态分配不支持星火流( 1.6.1版本)。但在2.0.0中是固定的
根据这个问题中的PDF，它说应该有一个名为spark.streaming.dynamicAllocation.enabled=true的配置字段，但是我在文档中没有看到这个配置。
有谁能确认一下，
我不能为1.6.1版本中的星火流启用动态资源分配吗？
它是否可在星火2.0.0中使用。如果是，应该设置什么配置(spark.streaming.dynamicAllocation.enabled=
浏览 0提问于2016-12-22得票数 7
我创建了一个简单的golang Apache光束管道，它在DirectRunner上工作得很好。我尝试使用以下命令将其部署到Spark集群上：./bin/spark-submit --master=spark://vm:7077 main.go --runner=SparkRunner --job_endpoint=localhost:8099 --artifact_endpoint=localhost:8098 --environment_type=LOOPBACK --output=/tmp/output
在提交应用程序之前，我使用以下命令运行了job_endpoint：
./gradle
浏览 71提问于2021-03-06得票数 0
回答已采纳
1回答
我正在运行一个Spark作业(从Spark notebook)，使用带有选项的动态分配
"spark.master": "yarn-client",
"spark.shuffle.service.enabled": "true",
"spark.dynamicAllocation.enabled": "true",
"spark.dynamicAllocation.executorIdleTimeout": "30s",
"spark.dynamicAlloc
浏览 14提问于2016-08-31得票数 1
当我想要发送一条AVRO消息，其中包含一个类型为long的字段时，我得到了以下错误：
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 61
Caused by: java.lang.ClassCastException: java.lang.Long cannot be cast to org.joda.time.DateTime
我使用Confluent 3.2.0和Apache Spark 2.2.0。此错误在Spark Job
浏览 36提问于2017-12-18得票数 5
回答已采纳
我已经设置了星火结构化流(Spark2.3.2)来阅读Kafka (2.0.0)。如果消息在星火流作业启动前进入主题，则无法从主题开始使用。这种预期的星火流行为是否忽略了最初运行Stream作业之前生成的Kafka消息(即使使用.option("stratingOffsets"，“最早”))？
复制步骤
在开始流作业之前，创建test主题(单个代理、单个分区)并生成主题的消息(在我的示例中有3条消息)。
使用以下命令启动spark：spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.
浏览 1提问于2019-06-19得票数 8
回答已采纳
我正在使用现有的多级分区拼图文件作为源处理结构化spark流。我在使用它的时候遇到了以下问题。
启动spark流作业以从特定分区读取数据，而不是从头开始。
假设我们观察到分区year=2018/month=10/hour=10存在数据质量问题。现在假设我已经通过替换正确的文件更正了到目前为止的数据。
现在的问题是如何从这一天开始重新处理数据，而不是从头开始？因为在结构化流传输中，比方说我使用文件流作为源，它将加载所有文件，我想忽略一些文件。这里我还需要删除我的检查点目录，因为它有截止日期的偏移量。
浏览 3提问于2019-10-14得票数 0
我有一个非常简单的火花流作业运行在本地独立模式。有一个客户接收器，它从数据库中读取，并将其传递给打印总数的主作业。不是一个实际的用例，但我是在四处游玩学习。问题是作业会永远陷入困境，逻辑非常简单，所以我认为它既不做任何处理，也不做内存问题。奇怪的是，如果我停止作业，突然在日志中我看到作业执行的输出和其他支持的作业跟随！有人能帮我理解这是怎么回事吗？
val spark = SparkSession
.builder()
.master("local[1]")
.appName("SocketStream")
.getOrCreate()
va
浏览 1提问于2017-06-05得票数 1
回答已采纳
我有一个简单的星火作业，它将数据流到Delta表。这张表很小，没有分区。
创建了许多小的拼花文件。
按照文档()中的建议，我添加了一个压缩作业，每天运行一次。
val path = "..."
val numFiles = 16
spark.read
.format("delta")
.load(path)
.repartition(numFiles)
.write
.option("dataChange", "false")
.form
浏览 5提问于2021-08-12得票数 4
下面是我的POM文件。我正在用aws kinesis写一篇spark streaming
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId&g
浏览 3提问于2017-04-27得票数 5
我正在将管道从循环CI迁移到Github操作，我觉得有点奇怪，我只能运行一次作业，而不是创建一个作业，然后从工作流部分调用它，这样就可以多次调用一个作业，而不会重复作业中的命令/脚本。
我的管道将代码推送到三个环境中，然后对每个环境运行灯塔扫描。在循环ci中，我有一个任务要将代码推送到env，还有一个任务要运行灯塔。然后，在我的工作流部分，我只调用3次作业，将env作为参数传递。我是遗漏了什么，还是在github行动中没有办法做到这一点？我是否只需要在每项工作中写出3次命令？
浏览 3提问于2021-12-03得票数 2
我使用streaming insert在Big Query Table中一次加载一行。这是代码 def insertBigQueryTable( tableName:String , datasetName:String, rowContent :java.util.Map[String, Object]): Unit = {
val bigquery = BigQueryOptions.getDefaultInstance.getService
try {
val tableId = TableId.of(datasetName, tableName)
浏览 21提问于2021-07-02得票数 1
回答已采纳
我是Spark的新手，所以任何建议，无论是关于相关工具或针对我的用例建议的设计更改，都将不胜感激。
我目前的情况是，我有几百万个独立的Spark作业，运行时间不是很长(平均几秒钟)，我使用Livy以批处理模式提交它们。问题是，初始化每个作业的Spark上下文所需的时间比运行作业本身所需的时间要长得多。因此，我的想法是初始化spark上下文的修复池，并使用它们来运行所有这些作业，而不是每次运行作业时都获得新的上下文。
问题是，我对Spark完全是个新手，不知道这是否可能，也不知道这是不是一个好主意。我尝试使用几个Livy会话，但是多个语句不能在一个Livy会话上同时执行，所以我被卡住了。
浏览 2提问于2021-09-06得票数 1
我有一个火花批处理代码(基本上是模型培训)，我从AWS集群用spark-submit执行。现在我希望能够每天在特定的时间启动这项工作。标准的方法是什么？我应该更改代码并在代码中添加调度吗？或者有什么方法来安排火花提交作业？或者我应该让它成为每24小时执行一次的星火流作业？(虽然我对一个特定的时间段感兴趣，即晚上11点到12点之间)
浏览 5提问于2016-11-10得票数 2
回答已采纳
1回答
默认情况下，Oozie共享库目录为Hive、Pig和Map提供了库。如果我想在Oozie上运行Spark作业，最好将Spark添加到Oozie的共享库中，而不是将它们复制到app的lib目录中。
我如何在Oozie的共享库中添加Spark (包括星火核心及其依赖项)？如有任何意见或答复，敬请见谅。
浏览 1提问于2015-06-01得票数 3
回答已采纳
1回答
过程组和工作之间有什么区别？如果我输入pr * | lpr，那么它既是一个进程组，也是一个作业？进程组ID和作业ID之间到底有什么区别？
编辑:我知道它看起来类似于一个工作和一个过程有什么区别？，但略有不同。而且，我没有从这个线程中理解这个概念。
浏览 0提问于2011-03-11得票数 11
回答已采纳
我有一个Kafka流作业(Spark version2.4.5)在kubernetes上运行，有一个驱动程序和4个executors.The，该作业的目的是消费来自PySpark主题的数据，并对它们进行处理，然后将最终数据加载到ElasticSearch。
问题：
有时，我们会遇到Kafka集群的问题。因此，PySpark流作业将不会获得任何要处理/消费的数据，即acceptable.However，即使在Kafka群集变得正常之后，PySpark流作业也不会消耗/处理来自Kafka topic.The作业的任何进一步记录
错误:在驱动程序或任何执行器上找不到错误。
其中一个executors
浏览 15提问于2021-11-25得票数 1
4回答
我有一个Spark Streaming作业，它一直在持续运行。如何优雅地停止作业？我已经阅读了在作业监视中附加关闭钩子并向作业发送SIGTERM的常用建议。
sys.ShutdownHookThread {
logger.info("Gracefully stopping Application...")
ssc.stop(stopSparkContext = true, stopGracefully = true)
logger.info("Application stopped gracefully")
}
这似乎是有效的，但看起来不是停止工作的
浏览 0提问于2015-09-15得票数 30
我有一个spark作业，正在将数据加载到dataproc集群中运行的BigQuery.The spark作业中。这是代码片段 df.write
.format("bigquery")
.mode(writeMode)
.option("table",tabName)
.save() 我已经在spark-submit命令的--jars参数中指定了spark bigquery依赖jar (spark-bigquery-with-dependencies_2.12-0.19.1.jar ) 当我运行代码时，我得到了以下
浏览 145提问于2021-08-02得票数 2
回答已采纳
1回答
我有一个连接到Apache Kafka数据流作为源的Spark 2.0.2结构化流作业。该工作从Kafka获取Twitter数据(JSON)，并使用CoreNLP为数据添加情感、词性标注等。它与local[*]主控一起工作得很好。然而，当我设置一个独立的Spark集群时，只有一个worker被用来处理数据。我有两个具有相同能力的工人。
提交作业时是否需要设置缺少的内容。我尝试在我的spark-submit命令中设置--num-executors，但是没有成功。
提前感谢您给我们指明了正确的方向。
浏览 0提问于2016-12-06得票数 1
2回答
我想将Nifi流文件发送到Spark，并在Spark中进行一些转换，然后再次将结果发送回Nifi，以便我能够在Nifi中进一步操作。我不想写流文件写到数据库或HDFS，然后触发火花作业。我想直接发送流文件到星火和接收结果直接从火花到尼菲。我试过在Nifi中使用ExecuteSparkInteractive处理器，但我被卡住了。任何例子都会有帮助。
浏览 1提问于2018-10-31得票数 8
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
