小编cyn*_*uit的帖子

我试图利用直接的kafka使用者（python中可用的新功能）从我在localhost：9092上运行的自定义Kafka Producer捕获数据。

我目前正在使用spark 1.6示例脚本提供的“ direct_kafka_wordcount.py”。

来源：https : //github.com/apache/spark/blob/master/examples/src/main/python/streaming/direct_kafka_wordcount.py

DOCS：http： //spark.apache.org/docs/latest/streaming-kafka-integration.html

我正在使用以下命令来运行程序：

    ~/spark-1.6.0/bin/spark-submit --jars 
    ~/spark-1.6.0/external/kafka-assembly/target/spark-streaming-kafka-assembly_2.10-1.6.0.jar           
direct_kafka_wordcount.py localhost:9092 twitter.live

不幸的是，我遇到一个奇怪的错误，无法调试。任何提示/建议将不胜感激。

py4j.protocol.Py4JJavaError: An error occurred while calling o24.createDirectStreamWithoutMessageHandler.
: org.apache.spark.SparkException: java.nio.channels.ClosedChannelException
        at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366)
        at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366)
        at scala.util.Either.fold(Either.scala:97)
        at org.apache.spark.streaming.kafka.KafkaCluster$.checkErrors(KafkaCluster.scala:365)
        at org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:222)
        at org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper.createDirectStream(KafkaUtils.scala:720)
        at org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper.createDirectStreamWithoutMessageHandler(KafkaUtils.scala:688)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
        at py4j.Gateway.invoke(Gateway.java:259)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:209)
        at java.lang.Thread.run(Thread.java:745)

我正在使用 Elasticsearch 2.2.0；但是，我真的很难尝试添加 geo_point 数据。事实上，地理编码数据被添加为字符串。

预期："geo":{"properties":{"location":{"type":"geo_point"}}}

实际："geo":{"properties":{"location":{"type":"string"}}}

我通过以下方式在python中添加数据：

from elasticsearch import Elasticsearch
es = Elasticsearch()

# ... 
es_entries['geo'] = { 'location': str(data['_longitude_'])+","+str(data['_latitude_'])}
# ...

es.index(index="geodata", doc_type="doc", body=es_entries)

有没有关于通过python添加geo_point数据的教程（这不像看起来那么简单）？

我在理解反向代理(即使用proxy_pass指令和给定的上游服务器)和301永久重定向之间的区别时遇到了一些困难.它们如何相似/不同？

反向代理

upstream backend {
    server backend1.example.com       weight=5;
    server backend2.example.com:8080;
}

server {
    location / {
        proxy_pass http://backend;
    }
}

HHTP重定向

Apache示例:http://www.inmotionhosting.com/support/website/htaccess/redirect-without-changing-url

NGINX示例:

server {
    listen 80;
    server_name domain1.com;
    return 301 $scheme://domain2.com$request_uri;
}

因此,似乎两种方法与最终用户的观点没有区别.我想确保有效的带宽使用,同时使用SSL.目前,应用服务器使用自己的自签名SSL证书和Nginx.将用户从标准Web托管公司(hostgator,godaddy等)托管的网站重定向到单独的服务器应用服务器的建议方法是什么？

如果这是重新发布的道歉,但我无法从mongodb文档示例中获得我想要的查询.

这是我的问题.我无法在单个查询中执行更新现有文档的array_field或添加新文档并使用初始值初始化array_field.

我可以使用一些条件逻辑的findOne(),并可能解决这个问题,但我认为mongodb有一个这个用例的实现

这是迄今为止的代码:

#data_json = JSON document to be added to collection

collection.update_one({"json_id":data_json["json_id"], "_dashbd_id_":dashboard_id},{{"$addToSet": {"array_field":keyword}},{"$setOnInsert":data_json}}, upsert=True)

我正在查询json_id和我的收藏中的_dashbd_id_.如果它存在,那么我打算将"keyword"添加到array_field.如果它不存在,请创建一个新文档data_json,其中包括array_field = [keyword]

任何提示和建议表示赞赏!