Pyspark: how to convert json strings in dataframe column

Question

Below is more or less direct python code that is functionally retrieved exactly as I want. The data schema for column I, which is filtered inside the data frame, is basically a json string.

However, I had to significantly increase the amount of memory for this, and I only work on one node. Using collection is probably bad, and creating all of this on a single node does not really use the distributed nature of Spark.

I would like to get a more effective Spark solution. Can someone help me massage the logic below to better use Spark? Also, as an educational point: please explain why / how updates make it better.

#!/usr/bin/env python # -*- coding: utf-8 -*- import json from pyspark.sql.types import SchemaStruct, SchemaField, StringType input_schema = SchemaStruct([ SchemaField('scrubbed_col_name', StringType(), nullable=True) ]) output_schema = SchemaStruct([ SchemaField('val01_field_name', StringType(), nullable=True), SchemaField('val02_field_name', StringType(), nullable=True) ]) example_input = [ '''[{"val01_field_name": "val01_a", "val02_field_name": "val02_a"}, {"val01_field_name": "val01_a", "val02_field_name": "val02_b"}, {"val01_field_name": "val01_b", "val02_field_name": "val02_c"}]''', '''[{"val01_field_name": "val01_c", "val02_field_name": "val02_a"}]''', '''[{"val01_field_name": "val01_a", "val02_field_name": "val02_d"}]''', ] desired_output = { 'val01_a': ['val_02_a', 'val_02_b', 'val_02_d'], 'val01_b': ['val_02_c'], 'val01_c': ['val_02_a'], } def capture(dataframe): # Capture column from data frame if it not empty data = dataframe.filter('scrubbed_col_name != null')\ .select('scrubbed_col_name')\ .rdd\ .collect() # Create a mapping of val1: list(val2) mapping = {} # For every row in the rdd for row in data: # For each json_string within the row for json_string in row: # For each item within the json string for val in json.loads(json_string): # Extract the data properly val01 = val.get('val01_field_name') val02 = val.get('val02_field_name') if val02 not in mapping.get(val01, []): mapping.setdefault(val01, []).append(val02) return mapping 

Answer 1

One possible solution:

 (df .rdd # Convert to rdd .flatMap(lambda x: x) # Flatten rows # Parse JSON. In practice you should add proper exception handling .flatMap(lambda x: json.loads(x)) # Get values .map(lambda x: (x.get('val01_field_name'), x.get('val02_field_name'))) # Convert to final shape .groupByKey()) 

The specified output specification, this operation is not entirely efficient (do you really need grouped values?), But still much better than collect .