и извлечь:
чномstructured_kafka_wordcount.py код,
Когда я делю строки на словаudf
как ниже,
my_split = udf(lambda x: x.split(' '), ArrayType(StringType()))
words = lines.select(
explode(
my_split(lines.value)
)
)
предупреждение будет продолжать показывать:
ПРЕДУПРЕЖДЕНИЕ CachedKafkaConsumer: CachedKafkaConsumer не работает в UninterruptibleThread. Может зависать, когда методы CachedKafkaConsumer прерываются из-за KAFKA-1894
С другой стороны, когда я делю строки на словаpyspark.sql.functions.split
все работает хорошо.
words = lines.select(
explode(
split(lines.value, ' ')
)
)
Почему это произошло и как исправить предупреждение?
Вот код, который я пытаюсь выполнить на практике:
pattern = "(.+) message repeated (\\d) times: \\[ (.+)\\]"
prog = re.compile(pattern)
def _unfold(x):
ret = []
result = prog.match(x)
if result:
log = " ".join((result.group(1), result.group(3)))
times = result.group(2)
for _ in range(int(times)):
ret.append(log)
else:
ret.append(x)
return ret
_udf = udf(lambda x: _unfold(x), ArrayType(StringType()))
lines = lines.withColumn('value', explode(_udf(lines['value'])))