Questions tagged [apache-spark-sql]

Apache Spark SQL is a tool for "SQL and structured data processing" on Spark, a fast and general-purpose cluster computing system. It can be used to retrieve data from Hive, Parquet etc. and run SQL queries over existing RDDs and Datasets.

Filter by
Sorted by
Tagged with
1 vote
1 answer
8 views

How to set partition size for date field in Databricks

I have timestamp field source_timestamp which is used a lot in the queries to the delta table. I would like to partition by this column with size of the partition = 1 month. How to achieve that? If I ...
archjkeee's user avatar
0 votes
0 answers
8 views

Replacing tables in SparkSQL when using PARQUET

I'm quite new to Spark and I'm trying to use PARQUET to create a table to resolve an issue with a spark runtime - the error message says the following: The depth of view 'foobar' exceeds the maximum ...
tayyib_saddique's user avatar
0 votes
1 answer
29 views

Generating weekly date column from two dates columns

I have a dataset which has multiple column, the columns that I'm interested in are below ID start_date end_date 1 1999-04-07 2013-09-15 1 2015-08-15 2017-02-01 2 2018-05-04 2020-04-02 8 ...
MS128's user avatar
  • 1
0 votes
0 answers
22 views

How can I perform the join among 7 tables (2 very large,1medium and 4small tables) efficiently in Pyspark

I have 7 iceberg tables and I'm reading them as dataframe in databricks using the below code `df1 = spark.read.format('iceberg').load(path1)` #row count 250million `df2 = spark.read.format('iceberg')....
Gorib_coder's user avatar
2 votes
1 answer
35 views

How I can define a window in spark sql. With fixed size; start with it's min value; non-overlapping

I need to define a window whose range depends on the value, but the range is fixed( for example 3). But the starting value of the window depends on the smallest possible value. And each window does ...
Smokeriu's user avatar
  • 167
-3 votes
1 answer
36 views

How to optimize a left join of 2 large tables having 300 gb data each and one table having 250 gb data [closed]

I have 3 tables, where 2 of the tables are built up in 2 different directories say s1/ and s2/ on AWS as parquet files having 300 files per directory. This distribution is based on a key which varies ...
codingprabhu's user avatar
0 votes
0 answers
26 views

How to optimise a left join where both the tables are quite big (1 tb size and other 250gb)

I have 8 tables, out of which one is 1 TB and other 7 are roughly around 270 GB. Each of the 7 tables needs to be left joined with the first table to get all the columns of first table and an extra ...
codingprabhu's user avatar
0 votes
0 answers
10 views

Pyspark MapOutputTrackerMaster Error while Running Dataproc Batch

In my pyspark serverless batch output log can see this warning ERROR MapOutputTrackerMaster: Asked to update map output for unknown shuffle 16 ERROR MapOutputTrackerMaster: Asked to update map output ...
Help_me_a_bit's user avatar
1 vote
1 answer
27 views

Difference between Pivot or Group By in SQL

I have data that looks as follows: | OmgevingID | AdministratieKantoorID | WerkgeverID | AdministratieID | JaarID | VolgnummerRun | PersoneelsnummerVerloning | Periode | Component | Tabel | Datum |...
Psychotechnopath's user avatar
1 vote
0 answers
25 views

Spark Hive Integration issue with eclipse

i am using cloudera-quickstart-vm-5.12.0-0-virtualbox-disk1 for my BigData practise. i was trying to integrate spark and hive using a scala code. this scala code is being written in windows eclipse. ...
Rajarshi Maity's user avatar
0 votes
0 answers
37 views

Ambiguity in definition of Spark "Column"

I'm working through this tutorial on manipulating data frame Column objects. The doc string for the Column class shows a Columns columnar data and heading (Annex A below). In contrast, the doc ...
user2153235's user avatar
1 vote
1 answer
35 views

Convert columns values to columns in spark

I have a column hit with schema as |-- hit: array (nullable = true) | |-- element: struct (containsNull = true) | | |-- key: string (nullable = true) | | |-- value: struct (nullable ...
Sk2415's user avatar
  • 51
0 votes
0 answers
38 views

Delta lake with Glue Catalog as metastore

We are using Spark (v3.3.0) SQL to consume data from a delta lake (v2.2.0), running on top of S3. The Spark jobs are run using Glue jobs and also EMR cluster. We want our code to refer to the delta ...
Subodh Shetty's user avatar
0 votes
1 answer
38 views

Pyspark Dataframe Column Arithmatic Operations

I have a dataframe in pysprak that has two columns, Flight Origin and Flight Destination. Origin Destination LAX JFK LHR LAX ATL DFW JFK LHR ATL LHR What I want to do using Pyspark dataframes ...
Ammad's user avatar
  • 3
1 vote
1 answer
69 views

Aggregate Data based on date difference range in Spark

I have an scenario where I need to aggregate data based on the Date difference of within two days range and aggregated date should have the oldest date as show in output. Input Dataframe InputDate ...
manmur's user avatar
  • 19
0 votes
1 answer
26 views

CTE query to generate new rows based on existing column in spark SQL

I have created a table(trial). create table trial( started int, ended int ); insert into trial values(1, 7), (5, 6), (9, 14); -- select * from trial; Now I want to generate new rows for each ...
The Flash's user avatar
0 votes
0 answers
21 views

Angle bracket notation in PySpark's Column<'FieldNameString'>

The PySpark documentation for the Column class has formatting issues that make some of the examples hard to read. The Annex below contains its doc string, which to me seems clearer. It shows typical ...
user2153235's user avatar
0 votes
1 answer
44 views

Range Join in PySpark / Spark SQL

I'm new to Databricks and Spark, and I have a query that seems suitable for range join optimization. However, I'm struggling with selecting the right bin size, even after reading the Azure and ...
Inês Silva's user avatar
0 votes
1 answer
25 views

How to create a new column in dataframe whose value is derived from other columns of the dataframe

I have a dataframe which has columns a,b. I want to create, in same dataframe, another column whose value (for each row) should be a*b. How do I do that? I tried few examples but none are working ...
Manu Chadha's user avatar
  • 15.6k
0 votes
1 answer
20 views

pyspark & iceberg: `update *` not working in `merge into`?

I run pyspark on AWS EMR studio MERGE INTO iceberg_catalog.staging.tbl AS t USING tempview AS s ON t.number = s.number WHEN MATCHED THEN UPDATE * WHEN NOT MATCHED THEN INSERT * but got pyspark.sql....
BAE's user avatar
  • 8,570
0 votes
1 answer
24 views

How to read .csv and .csv.gz file with all executors in EMR cluster

I am using aws emr service (emr 6.9.0) with one core node and 8 executors of instance type m5.2xlarge. My requirement is to execute read csv or csv.gz file through a spark scala job and write it into ...
Shan's user avatar
  • 31
0 votes
2 answers
41 views

Getting multivalue from the list of structs from nested structs

I have this schema of my data frame, the keys inside hosts are variables, basically, these are the hostnames and don't have any pattern Eg: ae0s01, aadasde002, sfvu75ae003 |-- hosts: struct (...
ankursingh1000's user avatar
0 votes
1 answer
33 views

Impelementing SCD type 2 in Palantir code repository

I have a requirement to implement the SCD type 2 logic using palantir code repository. If anyone has tried implementing it, can you please help me with the reference links please. I've implemented it ...
Arvind's user avatar
  • 192
1 vote
1 answer
25 views

stop spark dataframe distributing to cluster - it needs to stay on driver

we have a workload that computes on spark cluster workers (cpu intensive). The results are pulled back to the driver which has a large memory allocation to collect the results via RDD .collect() ...
DaManJ's user avatar
  • 393
1 vote
0 answers
40 views

Pyspark while joining and aggregating data

I am new to psypark, trying to write a code with joins and aggregations and then writing the output into a parquet. Code below: import pyspark.sql.functions as f a = spark.sql("""select ...
Paras Karandikar's user avatar
0 votes
1 answer
31 views

Spark worker node doesn't see added files

i have problem in running Java Spark application on my local machine. I'm running Spark master and 1 spark worker node as docker images with below config: version: '2.2' services: spark: image: ...
P. Kobędza's user avatar
1 vote
1 answer
26 views

How to update struct schema in a Spark sql table

My data lake has a spark-sql table (Spark 3.4.0) in Delta format. The table has a struct column and now I need to add a new field address to that struct column. What I have tried is: drop table if ...
user3692015's user avatar
1 vote
1 answer
25 views

How to assign a monotonically increasing number as a suffix to only duplicates in a column?

Input: pyspark dataframe firstname column has duplicates +---------+----------+--------+-----+------+------+ |firstname|middlename|lastname|id |gender|salary| +---------+----------+--------+-----+---...
Ishan Gote's user avatar
0 votes
1 answer
29 views

Default parallelism with Dynamic Allocation in spark

I am launching Spark application with dynamicAllocation enabled. I set the initialExecutors to be 10. Each executor has 4 cores. I set the maxExecutors to 80. As per my understanding when the spark ...
ASR's user avatar
  • 86
0 votes
0 answers
32 views

java.lang.NoClassDefFoundError: org/apache/spark/sql/internal/connector/SupportsStreamingUpdate

query = info_df_fin.selectExpr("to_json(struct(*)) AS value")\ .writeStream\ .format("kafka")\ .option("checkpointLocation", "...
Shaheer Shah's user avatar
1 vote
1 answer
50 views

PySpark DataFrame creation is throwing PySparkTypeError

I am new in PySpark and am trying to create a simple dataFrame from an array or dictionary and in both cases they are throwing the same exception. I have tired creating a dataframe from .csv files ...
Mojdeh Ebrahimi's user avatar
1 vote
2 answers
117 views

Spark SQL - Insert only to 2 columns in a table

I have following table created in spark SQL create table if not exists my_db.fire_service_calls_tbl( CallNumber integer, UnitID string, IncidentNumber integer, ...
LCJ's user avatar
  • 22.2k
0 votes
1 answer
26 views

Date Time Error when applying UDF in Pyspark

Trying to convert the string month in a pyspark dataframe into a month number using a User Defined Function with datetime library. But there is an error when running the function. I created the ...
Pulsara Gunawardhana's user avatar
0 votes
1 answer
28 views

Unnest and split function returning error in pyspark SQL

I have a query that works great in Presto format (when executing in Athena). However when I run the same query in AWS Glue (via a Python Spark.SQL dataframe), I get an error saying "...
cpljp's user avatar
  • 31
0 votes
1 answer
44 views

Is there a python library to convert schema from pyspark dataframe to avro schema?

I am building an application in which I am querying SQL Server table using pyspark. My plan is to push this to kafka, which will be consumed by Google Cloud Storage Sink Connector & saved in ...
downloaderfan's user avatar
0 votes
0 answers
19 views

Timestamp comparison in spark sql

I have below query in spark sql sparkSession.sql(f"( SELECT * FROM {ordersTable} \ WHERE lastUpdated >= current_timestamp() - INTERVAL 24 hours)") The date-time in ...
Amol Aggarwal's user avatar
0 votes
1 answer
32 views

Why are trees used in catalyst optimizer?

I want to know what is the need for trees in Catalyst optimizer and why are they used. What type of values do hold throughout the execution of a Spark application?
tru's user avatar
  • 55
1 vote
2 answers
52 views

Spark Array column - Find max interval between two values

I have a Scala Spark dataframe with the schema: root |-- passengerId: string (nullable = true) |-- travelHist: array (nullable = true) | |-- element: integer (containsNull = true)...
Abishek's user avatar
  • 777
0 votes
0 answers
11 views

Collecting data via windowing doesn't scale

I have flat table with a large number of rows. Some of these rows deemed to be important for some reason. I want to collect certain fields from rows around these rows in a window. I have a working ...
SDani's user avatar
  • 79
0 votes
1 answer
48 views

Max interval between occurrences of two values - Spark

I have a spark dataframe in the format below. I wanted to calculate the max travel streaks to other countries from India before returning back to India. I have created a flag to indicate if India was ...
Abishek's user avatar
  • 777
0 votes
0 answers
27 views

find id responsible for top 50 percent using running total

I have a pyspark dataframe ticket_zip. It has an id column called mch_id. It also has a column called qty that indicates the qty associated with the id column mch_id on each record. I would like to ...
user3476463's user avatar
  • 3,979
0 votes
1 answer
32 views

SPARK SQL evaluating string mathematical expressions from a column

I have a dataframe, which contains mathematical expressions in form of a string. Like Column A Column B Formula 10 20 x+y 10 20 x where x represents values from Column A and y from Column B. Using ...
Abhra's user avatar
  • 5
0 votes
1 answer
21 views

PySpark: Find specific value in a grouped data and mark entire group as different value

I have employee data like below. I want to group the below data by EMP_ID and if 'Status' of this grouped EMP_ID has the value 'Not Done' then entire 'overall_status' for the EMP_ID should be marked ...
sanju's user avatar
  • 9
0 votes
0 answers
27 views

PySpark: why does my application ignore one of the executors?

I have a cluster composed by a master and two executors. In the spark master application I can see both executors are alive, but when I start a spark application only one of them is used, and the ...
Sara Munafò's user avatar
1 vote
0 answers
39 views

Does Spark optimize higher order chained transformations?

In the example below, if we create N columns that use sorted_array, will Spark sort the original array N times or WholeStageCodegen will optimize it? from pyspark.sql import functions as F _data = [ ...
boyangeor's user avatar
  • 391
0 votes
0 answers
19 views

Condition for watermark interval for batch data in Spark Structured Streaming

I have two sources of data-itemDetails, admitDetails which come in batch mode. Sometimes this batch mode contains historical data as well. We are using Spark Structured Streaming which ingests data as ...
Rahul Kumar's user avatar
1 vote
2 answers
368 views

Pyspark - Reject Values based on multiple conditions

I have a dataframe with values: +--------------+---------------+------------------+------------+-----+ |registerNumber|joint_indicator|share_class_number|no_of_shares|mn_id| +--------------+-----------...
Muskan Makhija's user avatar
0 votes
1 answer
59 views

Joining two tables for delete using DeltaTable

I have an use case, where I have a file path which has "company_id" and "date". Have another s3 file where I need to delete data which is below year 2010 based on "company_id&...
Shasu's user avatar
  • 462
0 votes
2 answers
33 views

Issue while trying to filter records in spark dataframe column which has single quotes

I have spark dataframe df1. I want to filter based on failed_rules column which has single quotes. I tried below code but getting error. import spark.implicits._ val columns = Seq("failed_rules&...
Arvinth kumar's user avatar
0 votes
0 answers
30 views

Apache spark issue when spark.write is executed

I get below error while a spark.sql is executed and spark write is called. the sql does not haver any != operator. could not understand why it happens, anyone faced, could you please help? (pyspark....
Prabu's user avatar
  • 31

1
2 3 4 5
532