Pyspark 2.4.0, read avro from kafka with read stream – Python

by

You can include spark-avro package, for example using --packages (adjust versions to match spark installation):

bin/pyspark --packages org.apache.spark:spark-avro_2.11:2.4.0

and provide your own wrappers:

from pyspark.sql.column import Column, _to_java_column 

def from_avro(col, jsonFormatSchema): 
    sc = SparkContext._active_spark_context 
    avro = sc._jvm.org.apache.spark.sql.avro
    f = getattr(getattr(avro, "package$"), "MODULE$").from_avro
    return Column(f(_to_java_column(col), jsonFormatSchema)) 


def to_avro(col): 
    sc = SparkContext._active_spark_context 
    avro = sc._jvm.org.apache.spark.sql.avro
    f = getattr(getattr(avro, "package$"), "MODULE$").to_avro
    return Column(f(_to_java_column(col)))

Example usage (adopted from the official test suite):

from pyspark.sql.functions import col, struct


avro_type_struct = """
{
  "type": "record",
  "name": "struct",
  "fields": [
    {"name": "col1", "type": "long"},
    {"name": "col2", "type": "string"}
  ]
}"""


df = spark.range(10).select(struct(
    col("id"),
    col("id").cast("string").alias("id2")
).alias("struct"))
avro_struct_df = df.select(to_avro(col("struct")).alias("avro"))
avro_struct_df.show(3)

+----------+
|      avro|
+----------+
|[00 02 30]|
|[02 02 31]|
|[04 02 32]|
+----------+
only showing top 3 rows

avro_struct_df.select(from_avro("avro", avro_type_struct)).show(3)

+------------------------------------------------+
|from_avro(avro, struct<col1:bigint,col2:string>)|
+------------------------------------------------+
|                                          [0, 0]|
|                                          [1, 1]|
|                                          [2, 2]|
+------------------------------------------------+
only showing top 3 rows

More Related Contents:

  1. How to find median and quantiles using Spark
  2. Spark Dataframe distinguish columns with duplicated name
  3. How to change dataframe column names in pyspark?
  4. I can’t seem to get –py-files on Spark to work
  5. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  6. Spark RDD to DataFrame python
  7. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  8. Why is Apache-Spark – Python so slow locally as compared to pandas?
  9. Spark groupByKey alternative
  10. How to extract an element from a array in pyspark
  11. PySpark create new column with mapping from a dict
  12. Tuning parameters for implicit pyspark.ml ALS matrix factorization model through pyspark.ml CrossValidator
  13. Spark iteration time increasing exponentially when using join
  14. Create Spark DataFrame. Can not infer schema for type
  15. PySpark DataFrames – way to enumerate without converting to Pandas?
  16. Getting Spark, Python, and MongoDB to work together
  17. How to transform data with sliding window over time series data in Pyspark
  18. Using UDF ignores condition in when
  19. Create single row dataframe from list of list PySpark
  20. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  21. pyspark parse fixed width text file
  22. Concatenating string by rows in pyspark
  23. What is the equivalent to scala.util.Try in pyspark?
  24. When to cache a DataFrame?
  25. Pyspark – converting json string to DataFrame
  26. How to add suffix and prefix to all columns in python/pyspark dataframe
  27. Spark DAG differs with ‘withColumn’ vs ‘select’
  28. Pyspark changing type of column from date to string
  29. PySpark Evaluation
  30. What is the difference between spark-submit and pyspark?

Leave a Comment