- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在尝试使用 PySpark 和 Structured Streaming (Spark 2.3) 在两个 Kafka Stream 之间进行左外连接。
import os
import time
from pyspark.sql.types import *
from pyspark.sql.functions import from_json, col, struct, explode, get_json_object
from ast import literal_eval
from pyspark.sql import SparkSession
from pyspark.sql.functions import expr
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 pyspark-shell'
spark = SparkSession \
.builder \
.appName("Spark Kafka Structured Streaming") \
.getOrCreate()
schema_impressions = StructType() \
.add("id_req", StringType()) \
.add("ts_imp_request", TimestampType()) \
.add("country", StringType()) \
.add("TS_IMPRESSION", TimestampType())
schema_requests = StructType() \
.add("id_req", StringType()) \
.add("page", StringType()) \
.add("conntype", StringType()) \
.add("TS_REQUEST", TimestampType())
impressions = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "ip-ec2.internal:9092") \
.option("subscribe", "ssp.datascience_impressions") \
.load()
requests = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "ip-ec2.internal:9092") \
.option("subscribe", "ssp.datascience_requests") \
.option("startingOffsets", "latest") \
.load()
query_requests = requests \
.select(col("timestamp"), col("key").cast("string"), from_json(col("value").cast("string"), schema_requests).alias("parsed")) \
.select(col("timestamp").alias("timestamp_req"), "parsed.id_req", "parsed.page", "parsed.conntype", "parsed.TS_REQUEST") \
.withWatermark("timestamp_req", "120 seconds")
query_impressions = impressions \
.select(col("timestamp"), col("key").cast("string"), from_json(col("value").cast("string"), schema_impressions).alias("parsed")) \
.select(col("timestamp").alias("timestamp_imp"), col("parsed.id_req").alias("id_imp"), "parsed.ts_imp_request", "parsed.country", "parsed.TS_IMPRESSION") \
.withWatermark("timestamp_imp", "120 seconds")
query_requests.printSchema()
query_impressions.printSchema()
> root
|-- timestamp_req: timestamp (nullable = true)
|-- id_req: string (nullable = true)
|-- page: string (nullable = true)
|-- conntype: string (nullable = true)
|-- TS_REQUEST: timestamp (nullable = true)
>
> root |-- timestamp_imp: timestamp (nullable = true)
|-- id_imp: string (nullable = true)
|-- ts_imp_request: timestamp (nullable = true)
|-- country: string (nullable = true)
|-- TS_IMPRESSION: timestamp (nullable = true)
rawQuery = query_requests.join(query_impressions, expr("""
(id_req = id_imp AND
timestamp_imp >= timestamp_req AND
timestamp_imp <= timestamp_req + interval 5 minutes)
"""),
"leftOuter")
rawQuery = rawQuery \
.writeStream \
.format("parquet") \
.option("checkpointLocation", "/home/jovyan/streaming/applicationHistory") \
.option("path", "/home/jovyan/streaming").start()
print(rawQuery.status)
{'message': 'Processing new data', 'isDataAvailable': True, 'isTriggerActive': True} ERROR:root:Exception while sending command. Traceback (most recent call last): File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 1062, in send_command raise Py4JNetworkError("Answer from Java side is empty") py4j.protocol.Py4JNetworkError: Answer from Java side is empty
During handling of the above exception, another exception occurred:
Traceback (most recent call last): File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 908, in send_command response = connection.send_command(command) File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 1067, in send_command "Error while receiving", e, proto.ERROR_ON_RECEIVE) py4j.protocol.Py4JNetworkError: Error while receiving ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server (127.0.0.1:33968) Traceback (most recent call last):
File "/opt/conda/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 2910, in run_code exec(code_obj, self.user_global_ns, self.user_ns) File "", line 3, in print(rawQuery.status) File "/opt/conda/lib/python3.6/site-packages/pyspark/sql/streaming.py", line 114, in status return json.loads(self._jsq.status().json()) File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 1160, in call answer, self.gateway_client, self.target_id, self.name) File "/opt/conda/lib/python3.6/site-packages/pyspark/sql/utils.py", line 63, in deco return f(*a, **kw) File "/opt/conda/lib/python3.6/site-packages/py4j/protocol.py", line 328, in get_return_value format(target_id, ".", name)) py4j.protocol.Py4JError: An error occurred while calling o92.statusDuring handling of the above exception, another exception occurred:
Traceback (most recent call last): File "/opt/conda/lib/python3.6/site-packages/IPython/core/interactiveshell.py", line 1828, in showtraceback stb = value._render_traceback_() AttributeError: 'Py4JError' object has no attribute '_render_traceback_'
During handling of the above exception, another exception occurred:
Traceback (most recent call last): File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 852, in _get_connection connection = self.deque.pop() IndexError: pop from an empty deque
# Example:
# spark.master spark://master:7077
# spark.eventLog.enabled true
# spark.eventLog.dir hdfs://namenode:8021/directory
# spark.serializer org.apache.spark.serializer.KryoSerializer
spark.driver.memory 15g
# spark.executor.extraJavaOptions -XX:+PrintGCDetails -Dkey=value -Dnumbers="one two three"
ERROR:root:Exception while sending command. Traceback (most recent call last): File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 1062, in send_command raise Py4JNetworkError("Answer from Java side is empty") py4j.protocol.Py4JNetworkError: Answer from Java side is empty
During handling of the above exception, another exception occurred:
Traceback (most recent call last): File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 908, in send_command response = connection.send_command(command) File "/opt/conda/lib/python3.6/site-packages/py4j/java_gateway.py", line 1067, in send_command "Error while receiving", e, proto.ERROR_ON_RECEIVE) py4j.protocol.Py4JNetworkError: Error while receiving
最佳答案
我解决了问题!基本上,该问题出于某种原因与 Jupyter Notebook 有关。我删除了前面代码的下一行:
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 pyspark-shell'
> spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 spark_structured.py
关于apache-spark - 结构化流错误 py4j.protocol.Py4JNetworkError : Answer from Java side is empty,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/50171431/
在C语言中,当有变量(假设都是int)i小于j时,我们可以用等式 i^=j^=i^=j 交换两个变量的值。例如,令int i = 3,j = 5;在计算 i^=j^=i^=j 之后,我有 i = 5,
我为以下问题编写了以下代码: 给定一个由 N 个正整数组成的序列 A,编写一个程序来查找满足 i > A[j]A[i](A[i] 的 A[j] 次方 > A[j] 的 A[i] 次方)。 我的代码通过
这个表达式是从左到右解析的吗?我试图解释解析的结果,但最后的结果是错误的。 int j=10, k=10; j+=j-=j*=j; //j=j+(j-=j*=j)=j+(j-j*j) k+=k*=
给定一个整数数组 A ,我试图找出在给定位置 j ,A[j] 从每个 i=0 到 i=j 在 A 中出现了多少次。我设计了一个如下所示的解决方案 map CF[400005]; for(int i=0
你能帮我算法吗: 给定 2 个相同大小的数组 a[]和 b[]具有大于或等于 1 的整数。 查找不相等的索引 i和 j ( i != j ) 使得值 -max(a[i]*b[i] + a[i] * b
每次用J的M.副词,性能显着下降。因为我怀疑艾弗森和许比我聪明得多,我一定是做错了什么。 考虑 Collatz conjecture .这里似乎有各种各样的内存机会,但不管我放在哪里M. ,性能太差了
假设一个包含各种类型的盒装矩阵: matrix =: ('abc';'defgh';23),:('foo';'bar';45) matrix +---+-----+--+|abc|defgh|23|+
是否有可能对于两个正整数 i 和 j,(-i)/j 不等于 -(i/j)?我不知道这是否可能......我认为这将是关于位的东西,或者 char 类型的溢出或其他东西,但我找不到它。有什么想法吗? 最
假设两个不同大小的数组: N0 =: i. 50 N1 =: i. 500 应该有一种方法可以获得唯一的对,只需将两者结合起来即可。我发现的“最简单”是: ]$R =: |:,"2 |: (,.N0)
我是 J 的新用户,我只是想知道 J 包中是否实现了三次样条插值方法? 最佳答案 我自己不熟悉,但是我确实安装了所有的包,所以 $ rg -l -i spline /usr/share/j/9.02
在 Q/kdb 中,您可以使用 ': 轻松修改动词,它代表每个优先级。它会将动词应用于一个元素及其之前的邻居。例如 =': 检查值对是否相等。在 J 中,您可以轻松折叠 /\ 但它是累积的,是否有成对
嗨,我有一个 4x4 双矩阵 A 1+2i 2-1i -3-2i -1+4i 3-1i -3+2i 1-3i -1-3i 4+3i 3+5i 1-2i -1-4i
刚刚发现 J 语言,我输入: 1+^o.*0j1 I expected the answer to be 0 ,但我得到了 0j1.22465e_16。虽然这非常接近于 0,但我想知道为什么 J 应该
这个问题在这里已经有了答案: With arrays, why is it the case that a[5] == 5[a]? (20 个答案) 关闭 3 年前。 我正在阅读“C++ 编程语言”
当第一行是 1, 1/2 , 1/3 ....这是支持该问题的图像。 是否存在比朴素的 O(n^2) 方法更有效的方法? 我在研究伯努利数时遇到了这个问题,然后在研究“Akiyama-Tanigawa
我写了一段Java代码,它在无限循环中运行。 下面是代码: public class TestProgram { public static void main(String[] args){
for (int i = n; i > 0; i /= 2) { for (int j = 0; j 0; i /= 2) 的第一个循环结果 O(log N) . 第二个循环for (int
如问题中所述,需要找到数组中 (i,j) 对的总数,使得 (1) **ia[j]** 其中 i 和 j 是数组的索引。没有空间限制。 我的问题是 1) Is there any approach w
for l in range(1,len(S)-1): for i in range(1,len(S)-l): j=i+l for X in N:
第二个for循环的复杂度是多少?会是n-i吗?根据我的理解,第一个 for 循环将执行 n 次,但第二个 for 循环中的索引设置为 i。 //where n is the number elemen
我是一名优秀的程序员,十分优秀!