我已经在Web服务器上提供了一个.gz文件，我想以流方式消费，并将数据插入Couchbase。.gz文件中只有一个文件，又包含一个每行JSON对象。
由于Spark没有HTTP接收器，所以我自己写了一个（如下所示）。我正在使用Couchbase火花连接器进行插入。但是，运行时，作业实际上并没有插入任何内容。我怀疑这是由于我对Spark的经验不足，而不知道如何开始和等待终止。如下所示，可以进行两个这样的地方。
接收者：
public class HttpReceiver extends Receiver<String> {
private final String url;
public HttpReceiver(String url) {
super(MEMORY_AND_DISK());
this.url = url;
}
@Override
public void onStart() {
new Thread(() -> receive()).start();
}
private void receive() {
try {
HttpURLConnection conn = (HttpURLConnection) new URL(url).openConnection();
conn.setAllowUserInteraction(false);
conn.setInstanceFollowRedirects(true);
conn.setRequestMethod("GET");
conn.setReadTimeout(60 * 1000);
InputStream gzipStream = new GZIPInputStream(conn.getInputStream());
Reader decoder = new InputStreamReader(gzipStream, UTF_8);
BufferedReader reader = new BufferedReader(decoder);
String json = null;
while (!isStopped() && (json = reader.readLine()) != null) {
store(json);
}
reader.close();
conn.disconnect();
} catch (IOException e) {
stop(e.getMessage(), e);
}
}
@Override
public void onStop() {
}
}
dataload ：
public void load(String url) throws StreamingQueryException, InterruptedException {
JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(1000));
JavaReceiverInputDStream<String> lines = ssc.receiverStream(new HttpReceiver(url));
lines.foreachRDD(rdd ->
sql.read().json(rdd)
.select(new Column("id"),
new Column("name"),
new Column("rating"),
new Column("review_count"),
new Column("hours"),
new Column("attributes"))
.writeStream()
.option("idField", "id")
.format("com.couchbase.spark.sql")
.start()
// .awaitTermination(sparkProperties.getTerminationTimeoutMillis())
);
// ssc.start();
ssc.awaitTerminationOrTimeout(sparkProperties.getTerminationTimeoutMillis());
}
评论的线条表明我对启动和终止工作的困惑。另外，如果有问题或可以改进的话，请随时发表评论。
使用Spark V2.1.0与Java。
编辑1 ：
还尝试了此实现：
lines.foreachRDD(rdd ->
couchbaseWriter(sql.read().json(rdd)
.select(new Column("id"),
new Column("name"),
new Column("rating"),
new Column("review_count"),
new Column("hours"),
new Column("attributes"))
.write()
.option("idField", "id")
.format("com.couchbase.spark.sql"))
.couchbase()
);
ssc.start();
ssc.awaitTermination();
但它抛出了IllegalStateException: SparkContext has been shutdown
11004 [JobScheduler] ERROR org.apache.spark.streaming.scheduler.JobScheduler - Error running job streaming job 1488664987000 ms.0
java.lang.IllegalStateException: SparkContext has been shutdown
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1910)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1981)
at org.apache.spark.rdd.RDD$$anonfun$fold$1.apply(RDD.scala:1088)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
at org.apache.spark.rdd.RDD.fold(RDD.scala:1082)
at org.apache.spark.sql.execution.datasources.json.InferSchema$.infer(InferSchema.scala:69)
编辑2 ：事实证明，编辑1的错误是由我关闭上下文的@PostDestruct方法引起的。我正在使用春天，豆子应该是单身人士，但是在工作结束之前，以某种方式导致它导致它破坏。我现在删除了@PostDestruct并进行了一些更改。以下内容似乎在起作用，但有开放的问题：
public void load(String dataDirURL, String format) throws StreamingQueryException, InterruptedException {
JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(1000));
JavaReceiverInputDStream<String> lines = ssc.receiverStream(new HttpReceiver(dataDirURL));
lines.foreachRDD(rdd -> {
try {
Dataset<Row> select = sql.read().json(rdd)
.select("id", "name", "rating", "review_count", "hours", "attributes");
couchbaseWriter(select.write()
.option("idField", "id")
.format(format))
.couchbase();
} catch (Exception e) {
// Time to time throws AnalysisException: cannot resolve '`id`' given input columns: [];
}
});
ssc.start();
ssc.awaitTerminationOrTimeout(sparkProperties.getTerminationTimeoutMillis());
}
打开问题：
- 随着时间的推移
AnalysisException: cannot resolve 'ID' given input columns: [];。这是我的接收器的问题吗？ 当文档已经存在时，任务失败，以下例外。就我而言，我只是想覆盖文档，如果存在，不要炸毁。
Lost task 1.0 in stage 2.0 (TID 4, localhost, executor driver): com.couchbase.client.java.error.DocumentAlreadyExistsException at com.couchbase.client.java.CouchbaseAsyncBucket$13.call(CouchbaseAsyncBucket.java:475)
回答我自己的问题，这是我终于有效的情况，没有任何例外：
public void load(String dataDirURL, String format) throws InterruptedException {
JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(1000));
JavaReceiverInputDStream<String> lines = ssc.receiverStream(new HttpReceiver(dataDirURL));
ObjectMapper objectMapper = new ObjectMapper();
lines.foreachRDD(rdd -> {
JavaRDD<RawJsonDocument> docRdd = rdd
.filter(content -> !isEmpty(content))
.map(content -> {
String id = "";
String modifiedContent = "";
try {
ObjectNode node = objectMapper.readValue(content, ObjectNode.class);
if (node.has("id")) {
id = node.get("id").textValue();
modifiedContent = objectMapper.writeValueAsString(node.retain(ALLOWED_FIELDS));
}
} catch (IOException e) {
e.printStackTrace();
} finally {
return RawJsonDocument.create(id, modifiedContent);
}
})
.filter(doc -> !isEmpty(doc.id()));
couchbaseDocumentRDD(docRdd)
.saveToCouchbase(UPSERT);
}
);
ssc.start();
ssc.awaitTerminationOrTimeout(sparkProperties.getTerminationTimeoutMillis());
}