Questions tagged [pyspark]

Looking to utilize Python to split a JSON file into separate files based on the "transactionTypeName" found within the transactions.details. Each file should include all details starting from careperson to username. Here is an example of the JSON file afte ...

My approach involves utilizing a Spark dataframe to divide and retain data in a tabular layout. The content of my file appears like this - {"click_id": 123, "created_at": "2016-10-03T10:50:33", "product_id": 98373, "product_price": 220.50, "user_id": 1, " ...

I need to perform an outer join on two dataframes using Spark: Dataframe 1 columns: first_name, last, address Dataframe 2 columns: first_name, last_name, phone_number The keys for joining are first_name and df1.last==df2.last_name The desired schema fo ...

When executing this basic script in Pyspark from the terminal, it runs without any issues: import pyspark sc = pyspark.SparkContext() foo = sc.parallelize([1,2]) foo.foreach(print) However, running the same script in Rodeo results in an error message, w ...

Although a similar question has been briefly answered before, I couldn't add my additional query due to lack of minimum reputation. Hence, I am posting it here. I am trying to process Twitter data using Apache Spark + Kafka. I have created a pattern ...

Following up on a previous question thread, I want to thank @abiratis for the helpful response. We are now working on implementing the solution in our glue jobs; however, we do not have a static schema defined. To address this, we have added a new column c ...

I am working with a spark dataframe and have the following data: from pyspark.sql import SparkSession spark = SparkSession.builder.appName('').getOrCreate() df = spark.createDataFrame([(1, "a", "2"), (2, "b", "2"),(3, "c", "2"), (4, "d", "2"), ...

I need assistance with a script I have that generates a DataFrame, converts it to a CSV file, then sends it as an email attachment. The issue I'm facing is that the header and data are all in the first row, resulting in a CSV file with 60k columns and ...

I am currently in the process of transforming a python code into pyspark. My goal is to utilize fillna to replace any missing values with a value from another column within the same dataframe, specifically at index 0. Here is the original python code that ...

I've been struggling for a few days to get Spark working with my Jupyter Notebook and Anaconda setup. Here's how my .bash_profile is configured: PATH="/my/path/to/anaconda3/bin:$PATH" export JAVA_HOME="/my/path/to/jdk" export PYTHON_ ...

My goal is to identify the ultimate parent using Dir pandas, but I am facing a unique challenge where the graph structure may not align perfectly with my requirements. Here is the input data: Child Parent Class 1001 8888 A 1001 1002 D 1001 1002 ...

I am in need of constructing the final dataframe name dynamically based on the configuration (by joining final_df and suffix). Whenever I execute the code provided below, it throws an error - "SyntaxError: can't assign to operator". Strangely, if I re ...

Just starting out with PySpark and seeking advice on translating the following SAS code into PySpark: SAS Code: If ColA > Then Do; If ColB Not In ('B') and ColC <= 0 Then Do; New_Col = Sum(ColA, ColR, ColP); End; Else ...

Welcome to my first data frame that showcases player points Playername pid matchid points 0 Virat Kohli 10 2 0 1 Ravichandran Ashwin 11 2 9 2 Gautam Gambhir 12 2 1 3 Ravindra Jadeja 13 2 7 4 Amit Mishra 14 2 2 5 Mohammed Shami 15 2 2 6 ...

I am working with a pyspark.RDD that contains dates which I need to filter out. The dates are in the following format within my RDD: data.collect() = ["Nujabes","Hip Hop","04:45 16 October 2018"] I have attempted filtering t ...

I currently have a dataset structured like this: +-------+-------+-------+-------+ | Index |values_in_dicts | +-------+-------+-------+-------+ | 1 |[{"a":4, "b":5}, | | |{"a":7, "b":9}] | +----- ...

Is there a way to convert a Pyspark dataframe into a specific JSON format without the need to convert it to a Python pandas dataframe? Here is an example of our input Pyspark df: model year timestamp 0 i20 [2019, 2018 ...

I need to transform a dataframe with one row and multiple columns into a new format. Some columns contain single values, while others contain lists of the same length. My goal is to split each list column into separate rows while preserving the non-list co ...

I'm encountering an issue in spark.sql where I can't seem to convert a string to date format. Initially, when passing the raw string directly, it converts successfully. However, when attempting to store that value into a variable and use it as an argumen ...

Having trouble reading multiple csv files from various folders from pyspark.sql import * spark = SparkSession .builder .appName("example") .config("spark.some.config.option") .getOrCreate() folders = List(& ...

I am dealing with a DataFrame containing various combinations of batches, inputs, and outputs. My goal is to reintegrate their "unique combinations" back into the DataFrame. Here's a simplified snapshot of the data: Batch Output Input 1 A X ... (d ...

Hello, I am a newcomer to PySpark and currently grappling with a challenge that needs solving. I have the task of merging three columns based on the values in a fourth column: Let's consider an example table layout like this: store car color cyli ...

After transitioning from Python to Scala, I found myself struggling with the conversion of a program. Specifically, I encountered difficulties with two lines of code pertaining to creating an SQL dataframe. The original Python code: fields = [StructField ...

I've been working on some python spark code and I'm wondering how to pass a variable in a spark.sql query. q25 = 500 Q1 = spark.sql("SELECT col1 from table where col2>500 limit $q25 , 1") Unfortunately, the code above doesn't seem to work as i ...

I'm facing challenges while working with pyspark dataframes. One of the tasks involves handling a column named eventkey, which is formed by combining elements such as account_type, counter_type, and billable_item_sid. To accomplish this, I've developed a f ...

Currently, I am faced with the task of reading and converting a csv file that contains both json and non-json columns. After successfully reading the file and storing it in a dataframe, the schema appears as follows: root |-- 'id': string (null ...

Currently, I am utilizing python with pyspark for my projects. For testing purposes, I operate a standalone cluster on docker. I found this repository of code to be very useful. It is important to note that before running the code, you must execute this ...

Recently, I came across findspark and was intrigued by its potential. Up until now, my experience with Spark was limited to using spark-submit, which isn't ideal for interactive development in an IDE. To test it out, I ran the following code on Window ...

Encountering an issue while trying to read a large 6GB single-line JSON file: Error message: Job aborted due to stage failure: Task 5 in stage 0.0 failed 1 times, most recent failure: Lost task 5.0 in stage 0.0 (TID 5, localhost): java.io.IOException: Too ...

When it comes to using pyspark, I have a routine of executing the following code in jupyter. Do you think this process is always required? import findspark findspark.init('/opt/spark2.4') import pyspark sc = pyspark.SparkContext() ...

I have a JSON file with a dictionary that resembles the following: "object1":{"status1":388.646233,"status2":118.580561,"status3":263.673222,"status4":456.432483} I want to extract status1, status2, status ...

As a beginner in AWS Glue and Pyspark, I'm facing some challenges with a transformation task. My issue involves working with two DynamicFrames; one contains values in a specific column that need to be added as a new column in the other DynamicFrame. The va ...

Currently, I am in the process of fine-tuning the parameters of an ALS matrix factorization model that utilizes implicit data. To achieve this, I am utilizing pyspark.ml.tuning.CrossValidator to iterate through a parameter grid and identify the optimal mod ...

I'm a beginner in PySpark and to sum it up: I am faced with the challenge of reading a parquet file and using it with SPARK SQL. Currently, I have encountered the following issues: When I read the file with schema, it shows NULL values - using spark. ...

I am working with an RDD that contains two variables ID and time. The time variable is in the format of datetime.datetime. Here is a snapshot of the first few rows of the RDD data: [[41186, datetime.datetime(2014, 3, 1, 20, 48, 5, 630000)], [32036, date ...

I have been attempting to achieve a specific result in pyspark 2.4, but I am unsure of the best approach. My goal is to take a nested array and use it as the default value for a new column in my dataframe. from pyspark.sql import SparkSession from pyspar ...

The previous data set has at minimum two columns named os_first and os_last, and I want to include a new column called 'to_ios' Below are the criteria for assigning values: undefined if either os_first or os_last is empty 1 if 'os_las ...

Currently, I am learning from a tutorial that demonstrates the usage of the following code snippet to load data and perform simple data processing: from pyspark import SparkConf, SparkContext conf = SparkConf().setMaster("local").setAppName(&quo ...

I am currently facing an issue with importing my Excel file into PySpark on Azure-DataBricks machine so that I can convert it to a PySpark Dataframe. However, I am encountering errors while trying to execute this task. import pandas data = pandas.read_exc ...

import logging import findspark from cassandra.cluster import Cluster from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType def setup_keyspace(session): session.execute(""" CREA ...

I encountered an issue when calling a currency conversion function I created in Databricks. The error message is as follows: I attempted to resolve the problem with the following code snippet from pyspark.sql.functions import lit from pyspark.sql.function ...

Currently working on Python coding in Databricks using Spark 2.4.5. Trying to create a UDF that takes two parameters - a Dataframe and an SKid, where I need to hash all columns in that Dataframe based on the SKid. Although I have written some code for th ...

I am working with a csv file that has more than 96 million rows and seven columns. I need to perform a fuzzy search on one of the columns to find records with the highest similarity to a given input string. The file is being managed by spark, and I have lo ...

I need help with writing a PySpark query to count all the null values in a large dataframe. Here is what I have so far: import pyspark.sql.functions as F df_agg = df.agg(*[F.count(F.when(F.isnull(c), c)).alias(c) for c in df.columns]) df_countnull_agg.c ...

In need of converting JSON data into a tabular format for later transformation to parquet. Schema root |-- : string (nullable = true) sample data +----------------------------------------------+ +----------------------------------------------+ |{"d ...

We have encountered a situation where converting a JSON file to parquet results in the creation of numerous small parquet files. How can we prevent this from happening? What is the most effective and efficient approach to managing this transformation? Bel ...

I have a large Spark dataframe containing approximately 1 million rows. I am using pyspark and need to apply the box-cox transformation from the scipy library on each column of the dataframe. However, the box-cox function only accepts 1-d numpy arrays as i ...