Como "reduzir" várias tabelas json armazenadas em uma coluna de um RDD para uma única tabela RDD da forma mais eficiente possível
O acesso simultâneo para acrescentar linhas usando a união em um quadro de dados usando o código a seguir funcionará corretamente? Atualmente mostrando erro de tipo
from pyspark.sql.types import *
schema = StructType([
StructField("owreg", StringType(), True),StructField("we", StringType(), True)
,StructField("aa", StringType(), True)
,StructField("cc", StringType(), True)
,StructField("ss", StringType(), True)
,StructField("ss", StringType(), True)
,StructField("sss", StringType(), True)
])
f = sqlContext.createDataFrame(sc.emptyRDD(), schema)
def dump(l,jsid):
if not l.startswith("<!E!>"):
f=f.unionAll(sqlContext.read.json(l))
savedlabels.limit(10).foreach(lambda a: dump(a.labels,a.job_seq_id))
Suponha que sqlContext.read.json (l) leia um json e produza um RDD com o mesmo esquema
O padrão é que eu quero "reduzir" várias tabelas json armazenadas em uma coluna de um RDD para uma tabela RDD da forma mais eficiente possível.
def dump(l,jsid):
if not l.startswith("<!E!>"):
f=f.unionAll(sc.parallelize(json.loads(l)).toDF())
O código acima também não funcionará, pois sc.parallelize está sendo chamado pelos threads de trabalho. Daí como resolver este problema?