我需要将火花应用程序/作业提交到远程火花群集.我目前在我的机器上点火,主节点的IP地址作为yarn-client.顺便说一句我的机器不在群集中.我用这个命令提交我的工作
./spark-submit --class SparkTest --deploy-mode client /home/vm/app.jar
我将我的主人的地址硬编码到表单中的应用程序中
val spark_master = spark://IP:7077
但我得到的只是错误
16/06/06 03:04:34 INFO AppClient$ClientEndpoint: Connecting to master spark://IP:7077...
16/06/06 03:04:34 WARN AppClient$ClientEndpoint: Failed to connect to master IP:7077
java.io.IOException: Failed to connect to /IP:7077
at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:216)
at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:167)
at org.apache.spark.rpc.netty.NettyRpcEnv.createClient(NettyRpcEnv.scala:200)
at org.apache.spark.rpc.netty.Outbox$$anon$1.call(Outbox.scala:187)
at org.apache.spark.rpc.netty.Outbox$$anon$1.call(Outbox.scala:183)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.ConnectException: Connection refused: /IP:7077
或者如果我使用
./spark-submit --class SparkTest --master yarn --deploy-mode client /home/vm/test.jar
我明白了
Exception in thread "main" java.lang.Exception: When running with master …在Zeppelin中运行Spark bisecting kmmeans算法.
//I transform my data using the TF-IDF algorithm
val idf = new IDF(minFreq).fit(data)
val hashIDF_features = idf.transform(dbTF)
//and parse the transformed data to the clustering algorithm.
val bkm = new BisectingKMeans().setK(100).setMaxIterations(2)
val model = bkm.run(hashIDF_features)
val cluster_rdd = model.predict(hashIDF_features)
我总是得到这个错误:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 270.0 failed 4 times, most recent failure: Lost task 0.3 in stage 270.0 (TID 126885, IP): java.util.NoSuchElementException: key not found: 67
at scala.collection.MapLike$class.default(MapLike.scala:228)
at scala.collection.AbstractMap.default(Map.scala:58)
at …所以我有一张表，我想创建一个复合主键：一个分区键和几个集群列。然而，严格来说这些列并不总是填充的，因此某些行可能具有空值。Cassandra 允许这样做吗？具有空值的聚类列？
以下是教程中的测试命令：
./logstash -e 'input { stdin { } } output { stdout {} }'
以下是错误。
WARNING: Could not find logstash.yml which is typically located in $LS_HOME/config or /etc/logstash. You can specify the path using --path.settings. Continuing using the defaults
Could not find log4j2 configuration at path //usr/share/logstash/config/log4j2.properties. Using default config which logs to console
01:55:14.242 [main] FATAL logstash.runner - An unexpected error occurred! {:error=>#<ArgumentError: Path "/usr/share/logstash/data" must be a writable directory. It is not writable.>, :backtrace=>["/usr/share/logstash/logstash-core/lib/logstash/settings.rb:433:in `validate'", "/usr/share/logstash/logstash-core/lib/logstash/settings.rb:216:in `validate_value'", "/usr/share/logstash/logstash-core/lib/logstash/settings.rb:132:in …我有三个相同大小的RDD rdd1包含一个String标识符,rdd2包含一个向量并rdd3包含一个整数值.
基本上我想将这三个压缩在一起得到一个RDD,RDD[String,Vector,Int]但我不断得到不能用不等数量的分区压缩RDD.我怎样才能完全绕过拉链来做上述事情呢？
我正在尝试使用Spark对cassandra执行查询.系统获取特定值,我想在DB中找到包含该值的其他元素.但是,该值是用户生成的,它取决于查询(即,它不是预定义的,系统每次都会收到一个新的).因此它将在方法中的某处定义为:
val id = "ID"
我将要查询的colmn被DB视为TEXT,但实际上它是一个json输出,我必须在json字段中找到一个特定的值.因此,在spark命令中,我也在转换Json中每行的内容并遍历.
所以查询看起来像这样:
sc.cassandraTable("keyspace","table").where("some_restriction=random")
.filter(x=> (Json.parse(x.get[String]("content"))\"id")
.toString.contains(id))
这会引发错误:
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2055)
at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:341)
at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:340)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
at org.apache.spark.rdd.RDD.filter(RDD.scala:340)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:61)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:66)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:68)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:70)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:72)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:74)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:76)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:78)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:80)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:82)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:84)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:86)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:88)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:90)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:92)
at $iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:94)
at $iwC$$iwC$$iwC$$iwC.<init>(<console>:96)
at $iwC$$iwC$$iwC.<init>(<console>:98)
at $iwC$$iwC.<init>(<console>:100)
at $iwC.<init>(<console>:102)
at <init>(<console>:104)
at .<init>(<console>:108) …我使用 COPY/TO 创建了一个标准的 csv 文件，现在我想在另一个 cassandra 表中使用 COPY/FROM 上传内容。但是，我收到以下错误：
Failed to import 5000 rows: Error - field larger than field limit (131072), given up after 1 attempts
Exceeded maximum number of insert errors 1000
Failed to process 5000 rows; failed rows written to import_point_labeled_results.err
Exceeded maximum number of insert errors 1000
Processed: 0 rows; Rate: 0 rows/s; Avg. rate: 0 rows/s
0 rows imported from 0 files in 6.639 seconds (0 skipped).
csv 有大约 90 万个条目。是的，某些列内容大于 131072，但这是不可避免的，无法更改。如何解决此问题/绕过此问题？我真的很想这样做而不必重新启动整个集群。
编辑：
我尝试遵循更改./cqlshrc …
我有一个 Spark RDD，我想以有组织的方式对其条目进行排序。假设条目是一个包含 3 个元素的元组(name,phonenumber,timestamp)。我想首先根据 的值对条目进行排序phonenumber，然后根据 的值进行timestamp排序，同时尊重而不是更改基于phonenumber. （所以timestamp只根据phonenumber排序重新排列）。是否有 Spark 函数来执行此操作？
（我在 Scala 中使用 Spark 2.x）
当我声明我的 Go 环境时（即使用简单导出的 GOPATH 和 GOROOT）：
export GOROOT=/usr/lib/go-1.9/
export GOPATH=/my/workspace/go
当前终端正常识别变量，但如果我打开另一个终端窗口，这些变量没有设置，需要从头开始重新配置。
除了像编辑 .bashrc 文件这样的常见方法之外，是否有一种特定于 go 的方法来设置 go 环境变量的默认值？