How to convert a dictionary to a dataframe in pyspark? -splunktool (2023)

Last update :2022-08-08 10:55 a. m.
Technological knowledge: python

Now create a PySpark DataFrame from the Dictionary object and name it properties. In Pyspark, key and value types can be any Spark type that extends org.apache.spark.sql.types.DataType . Let's see how to extract key and values from PySpark DataFrame Dictionary column. Here I used PySpark's map transform to read the property values (MapType column). As I said at the beginning, PySpark does not have a dictionary type, instead it uses MapType to store the dictionary object, below is an example how I create a DataFrame MapType column using pyspark.sql.types.StructType. This displays the schema of the PySpark DataFrame and the result of the DataFrame. Note that dictionary column properties are represented as a map in the following schema.

In this article, I will explain how to manually create a PySpark DataFrame from Python.sayingand explain how to read Dict elements by key and some map operations using SQL functions. First let's create data with a list of Python Dictionary (Dict) objects, the following example has 2 columns of type String & Dictionary as{key:value,key:value}.

data dictionary = [ ('Jaime', {'hair':'negro','ojo':'Brown'}), ('Miguel', {'hair':'Brown','ojo':None}), ('Roberto', {'hair':'rojo','ojo':'negro'}), ('Washington', {'hair':'rojo','ojo':'gris'}), ('Jefferson', {'hair':'rojo','ojo':''})]

2._

df = spark.cryarDataFrame(data = data dictionary, schema = ["name","properties"])df.print layout()df.show(truncate =FALSE)

This shows thePySpark DataFrame Schema& Result of the DataFrame. Note that the dictionary columnpropertiesis represented asmapin the following scheme.

root | --name:chain(nullable =TRUE) | --properties:map(nullable =TRUE) | | --key: string | | --valeria:chain(valueContains Null =TRUE) + -- -- -- -- -- + -- -- -- -- -- -- -- -- -- -- -- -- -- -- - + | name | properties | + -- -- -- -- -- + -- -- -- -- -- -- -- -- -- -- -- -- - + |jaime| [eye -> brown, hair -> black] | |miguel| [eye -> , hair -> brown] | |roberto| [eye -> black, hair -> red] | |Washington| [eye -> grey, hair -> grey] | |jefferson| [eye -> , hair -> brown] | + -- -- -- -- -- + -- -- -- -- -- -- -- -- -- -- -- -- -- - +

Let's see how to extract key and values from PySpark DataFrame Dictionary column. Here I used the PySpark map transform to read the values fromproperties(Map type column)

d.f..rdd.map(lambda x: (x.name, x.properties["hair"], x.properties["ojo"]) ).aDF(["hair","ojo"]).show() + -- -- -- -- -- + -- -- - + -- -- - + | name | hair | eye | + -- -- -- -- -- + -- -- - + -- -- - + | Jaime | black | brown | | Miguel | brown | null | | Roberto | red | black | | Washington | gray | gray | | Jeferson | brown | | + -- -- -- -- -- + -- -- - + -- -- - +

6._

d.f.conColumna("hair", df.properties.take the object("hair"))\.conColumna("ojo", df.properties.take the object("ojo"))\.to tear down("properties")\ .show()df.conColumna("hair", df.properties["hair"])\ .conColumna("ojo", df.properties["ojo"])\ .to tear down("properties")\ .show()

suggestion: 2

old-fashioned:

Carolina do sul.parallelize([{"arg1":"","arg2":""}, {"arg1":"","arg2":""}, {"arg1":"","arg2":""}]).aDF()

new way:

ofspark.sqlimport lineofcollectionimport OrderedDictdef convert_to_row(d: dict) -> Row: returnfila( **orderly dictation(clean(D.elements())))Carolina do sul.parallelize([{ "arg1":"","arg2":""}, { "arg1":"","arg2":""}, { "arg1":"","arg2":""}])\.map(convert_to_line)\.toDF()

Then

{k1: v1, k2: v2...}

becomes

-- -- -- -- -- -- -- -- | coluna1 | col2 | | -- -- -- -- -- -- -- -- | | k1 | v1 | | k2 | v2 | -- -- -- -- -- -- -- -- jaja =list(map(lista, midic.elements())) df = spark.cryarDataFrame(hahahaha, ["col1","col2"])

The other answers work, but here's one more sentence that works well with nested data. It might not be the most efficient, but if you're creating a DataFrame from an in-memory dictionary, working with small datasets like test data, or using Spark incorrectly, efficiency shouldn't be an issue. Worry:

d = {any json compatible dictation}spark.read.json(sc.parallelize([json.dumps(d)]))

I had to modify the accepted answer to make it work for me on Python 2.7 with Spark 2.0.

ofcollectionsmatter orderly dictationofsparksqlmatter SparkSession,filaspark = (SparkSession.constructor.get or create())schema =type of structure([structure field('arg1',type of current(),TRUE),structure field('arg2',type of current(),TRUE)])dta = [{"arg1":"","arg2":""}, {"arg1":"","arg2":""}]dtaRDD = spark.spark context.parallelize(dta)\.map(lambdax:fila( **orderly dictation(clean(X.elements()))))dtaDF = spark.cryarDataFrame(dtaRdd, scheme)

Assuming your data is a structure and not a dictionary of strings, you can do

nuevodf = df.select(['df.arg1','df.arg2'])

suggestion: 3

This article shows how to convert a list of Python dictionaries to a DataFrame in Spark using Python. The following code snippets directly create the data frame using the function SparkSession.createDataFrame, create Create create Write Article draw Draw diagram post_add Start a conversation

Example list of dictionaries

data = [{"Category":'Category A',"I WENT":1,"Valentine":12h40}, {"Category":'Category B',"I WENT":2,"Valentine":30.10}, {"Category":'Category C',"I WENT":3,"Valentine":100,01}]

code snippet

from pyspark.sqlimport SparkSessionappName = "Python Example - PySpark Parsing Dictionary as Data Frame"master = "local"# Create Sparkpark Sessions = SparkSession.builder\ .appName(appName)\ .master(master)\ .getOrCreate ( )# Listdata = [ { "Category": 'Category A', "ID": 1, "Value": 12.40 }, { "Category": 'Category B', "ID": 2, "Value" : 30.10 } , { "Category" : 'Category C', "ID": 3, "Value": 100.01 }]# Create data framedf = spark.createDataFrame(data)print(df.schema)df.show( )

Following is the output of the above PySpark script.

session.py:340:UserWarning: infer schema dict is deprecated, use pyspark.sql.Row instead of warnings.to warn("Infer dict schema is deprecated", type of structure(List(structure field(Category, StringType, true),structure field(ID, LongType, true),structure field(Value, DoubleType, true))) + -- -- -- -- -- + -- - + -- -- -- + | Category | identification | Value | + -- -- -- -- -- + -- - + -- -- -- + | Category A |1|12.4|| Category B |2|30.1|| Category C |3|100,01| + -- -- -- -- -- + -- - + -- -- -- +

However, there is a caveat:

Notice: inference schemeofdict is deprecated, use pyspark.sql.filainstead of

7._

scheme =type of structure([structure field('Category',type of current(),FALSE),structure field('I WENT',integer type(),FALSE),structure field('Valentine',decimal type(scale =2),TRUE)])

1._

type of structure(List(structure field(Category,type of current,TRUE),structure field(I WENT,long type,TRUE),structure field(Valentina,double type,TRUE)))

However, there is a caveat:

Notice: inference schemeofdict is deprecated, use pyspark.sql.filainstead of

code snippet

from pyspark.sqlimport SparkSession, RowappName = "Exemplo Python - Dicionário PySpark Parse como DataFrame"master = "local"# Create Sparkpark Sessions = SparkSession.builder\ .appName(appName)\ .master(master)\ .getOrCreate()# Listdata = [{ "Categoria": 'Categoria A', "ID": 1, "Valor": 12,40 }, { "Categoria": 'Categoria B', "ID": 2, " Valor": 30,10 }, { "Category": 'Category C', "ID": 3, "Value": 100.01 }]# Create Data framedf = spark.createDataFrame([Row( ** i) for i in data])print(df .schema)df .show()

scheme =type of structure([structure field('Category',type of current(),FALSE),structure field('I WENT',integer type(),FALSE),structure field('Valentine',decimal type(scale =2),TRUE)])

suggestion: 4

The type of key-value pairs can be customized with parameters (see below). This method should only be used if the resulting pandas dataframe is expected to be small, as all data is loaded into controller memory. Convert the DataFrame to a dictionary. Determines the type of dictionary values.

>>> df = pd.data frame({'col1': [1,2], ...'col2': [0,5,0,75] }, ...index = ['fila1','fila2'], ...columns = ['col1','col2']) >>> dfcol1 col2fila11 0,50fila22 0,75

>>> df_dict = df.to_dict() >>>clean([(key,clean(values.elements()))forkey, valuesemdf_dict.elements()])[('col1', [('fila1',1), ('fila2',2)]), ('col2', [('fila1',0,5), ('fila2',0,75)])]

>>> df_dict = df.to_dict('series') >>>clean(df_dic.elements())[('col1', fila11fila22 Name: col1,kind of d: int64), ('col2', fila10,50fila20,75 Name: col2,kind of d: flotar64)]

>>> df_dict = df.to_dict('to divide') >>>clean(df_dic.elements())[('columns', ['col1','col2']), ('data', [ [1...,0,75] ]), ('index', ['fila1','fila2'])]

>>> df_dict = df.to_dict('records') >>> [clean(values.elements())forvaluesemdf_dict] [ [('col1',1...), ('col2',0,5)], [('col1',2...), ('col2',0,75)]]

>>> df_dict = df.to_dict('index') >>>clean([(key,clean(values.elements()))forkey, valuesemdf_dict.elements()])[('fila1', [('col1',1), ('col2',0,5)]), ('fila2', [('col1',2), ('col2',0,75)])]

suggestion: 5

11/03/2022

Add the variable's JSON content to a list.

matterscale.collection.mutable.ListBuffervalor json_content1 ="{'json_col1': 'hora', 'json_col2': 32}"valor json_content2 ="{'json_col1': 'hello', 'json_col2': 'world'}"erajson_seq =novo ListBuffer[Chain]()json_seq += json_content1json_seq += json_content2

Create a Spark dataset from the list.

valor json_ds = json_seq.toDS()

To usechispa.read.jsonto analyze the Spark dataset.

valor df = spark.read.json(json_ds)screen(df)

Add the variable's JSON content to a list.

matterscale.collection.mutable.ListBuffervalor json_content1 ="{'json_col1': 'hora', 'json_col2': 32}"valor json_content2 ="{'json_col1': 'hello', 'json_col2': 'world'}"erajson_seq =novo ListBuffer[Chain]()json_seq += json_content1json_seq += json_content2

Create a Spark dataset from the list.

valor json_ds = json_seq.toDS()

To usechispa.read.jsonto analyze the Spark dataset.

valor df = spark.read.json(json_ds)screen(df)

matterscale.collection.mutable.ListBuffervalor json_content1 ="{'json_col1': 'hora', 'json_col2': 32}"valor json_content2 ="{'json_col1': 'hello', 'json_col2': 'world'}"erajson_seq =novo ListBuffer[Chain]()json_seq += json_content1json_seq += json_content2val json_ds = json_seq.toDS()val df = faísca.file.json(json_ds)Show(fd)

Select the JSON column from a DataFrame and convert it to an RDD of typeRDD[Line].

matterorg.apache.spark - spark.sql.functions._valor test_df =sequence(("1","{'json_col1': 'hora', 'json_col2': 32}","1,0"), ("1","{'json_col1': 'hello', 'json_col2': 'world'}","1,0")).aDF("line number","json","symbolic")val fila_rdd = test_df.select(column("json")).rdd // Select only the JSON column and convert it to RDD.

Select the JSON column from a DataFrame and convert it to an RDD of typeRDD[Line].

matterorg.apache.spark - spark.sql.functions._valor test_df =sequence(("1","{'json_col1': 'hora', 'json_col2': 32}","1,0"), ("1","{'json_col1': 'hello', 'json_col2': 'world'}","1,0")).aDF("line number","json","symbolic")val fila_rdd = test_df.select(column("json")).rdd // Select only the JSON column and convert it to RDD.

ConverterRDD[Line]aRDD[String].

val string_rdd = line_rdd.map(_.cadena mk(","))

trending technology

android × 15970
angular × 18982
API × 5999
css × 155656
html × 23420
java × 29699
javascript × 58592
json × 19745
php × 23600
python × 454736
react x 27451
sql × 2974
typed × 8420
xml × 3800