Подсчет количества запятых в датафрейме PySpark с исключением запятых, заключенных в двойные кавычки.

Вопрос или проблема

# Регулярное выражение для подсчета запятых, не находящихся в двойных кавычках
df = df.withColumn(
    "CommaCount",
    F.size(F.split(F.col("SingleColumn"), r',(?=(?:[^"]*"[^"]*")*[^"]*$)')) - 1
)

import re
str="30,kUsUO,6,18,97,42,"SAM,K,KARAN",lmhYK,49,aLaTA,51,34,3,49,75,39,pdwvW,54,7,63,12,25,26,SJ12u,rUFUV,34,xXBv3,XHtz4,r4Fyh,14,20,0jZL2,izrsC,44,K5Kw3,8,tcKu7,5,RPLcy,kg4IR,Kvs3p,lyG09,dJmZB,34,84,7,qED2y,8uNen,5,96,81,88,bGgqK,FAsIV,81,YXZ,PQR,"Flat No B1002, Balaji Whitefield society, sus road, pune,Mh",22,591213,LbAo7,21,18,text,,,,,"
print(str.count(',') - sum(s.count(',') for s in re.findall('"([^"]*)"', str)))

68

Ответ или решение

Для решения задачи подсчета количества запятых в строках столбца DataFrame PySpark, исключая запятые, находящиеся в двойных кавычках, можно использовать более эффективный подход, чтобы избежать долгих вычислений на больших объемах данных (например, 100 миллионов записей).

В данном случае, можно воспользоваться функциональными возможностями PySpark без сложных регулярных выражений. Рассмотрим следующий код, который выполняет необходимые вычисления:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# Инициализация SparkSession
spark = SparkSession.builder \
    .appName("Count Commas Excluding Quotes") \
    .getOrCreate()

# Примерные данные
data = [("30,kUsUO,6,18,97,42,\"SAM,K,KARAN\",lmhYK,49,aLaTA,51,34,3,49,75,39,pdwvW,54,7,63,12,25,26,SJ12u,rUFUV,34,xXBv3,XHtz4,r4Fyh,14,20,0jZL2,izrsC,44,K5Kw3,8,tcKu7,5,RPLcy,kg4IR,Kvs3p,lyG09,dJmZB,34,84,7,qED2y,8uNen,5,96,81,88,bGgqK,FAsIV,81,YXZ,PQR,\"Flat No B1002, Balaji Whitefield society, sus road, pune,Mh\",22,591213,LbAo7,21,18,text,,,,," ,)]
df = spark.createDataFrame(data, ["text"])

# Функция для подсчета запятых, не учитывая запятые внутри кавычек
def count_commas(text):
    # Сначала нужно подсчитать общее число запятых
    total_commas = text.count(',')
    # Затем считаем запятые внутри кавычек, используя регулярное выражение
    quoted_commas = sum(s.count(',') for s in re.findall(r'"([^"]*)"', text))
    # Возвращаем разность между общим числом запятых и запятыми внутри кавычек
    return total_commas - quoted_commas

# Создаем UDF для использования в DataFrame
from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

count_commas_udf = udf(count_commas, IntegerType())

# Применяем UDF к столбцу с текстом
df_with_commas = df.withColumn("CommaCount", count_commas_udf(F.col("text")))

# Вывод результата
df_with_commas.show(truncate=False)

Объяснение кода:

1. Импорт библиотек: Мы импортируем необходимые библиотеки pyspark и re.

2. Инициализация Spark: Создание SparkSession, необходимого для работы с DataFrame.

3. Создание DataFrame: Пример данных помещается в DataFrame с одним столбцом text.

4. Функция count_commas: Эта функция принимает строку, подсчитывает общее количество запятых в строке и количество запятых, находящихся в двойных кавычках, затем возвращает разницу.

5. Создание UDF: Мы создаем пользовательскую функцию (UDF), чтобы применять нашу функцию подсчета запятых к каждому элементу столбца в DataFrame.

6. Применение UDF: Мы добавляем новый столбец CommaCount в DataFrame, в котором хранится количество запятых, исключая запятые внутри кавычек.

7. Вывод результата: В конце выводим DataFrame с подсчитанным количеством запятых.

Этот подход должен быть более производительным, чем использование сложных регулярных выражений, особенно на больших наборах данных.