Как разрешить повторяющиеся имена столбцов при объединении двух фреймов данных в PySpark? - PullRequest
Новая
       12

Как разрешить повторяющиеся имена столбцов при объединении двух фреймов данных в PySpark?

0 голосов
/ 11 марта 2019

У меня есть файлы A и B, которые в точности совпадают.Я пытаюсь выполнить внутреннее и внешнее объединение этих двух информационных кадров.Поскольку у меня есть все столбцы как повторяющиеся столбцы, существующие ответы не помогли.Другие вопросы, которые я прошел, содержат одинаковые столбцы или два, моя проблема в том, что все файлы являются дубликатами друг друга: и в данных, и в именах столбцов.

Мой код: 

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql import DataFrameReader, DataFrameWriter
from datetime import datetime

import time

# @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

print("All imports were successful.")

df = spark.read.orc(
    's3://****'
)
print("First dataframe read with headers set to True")
df2 = spark.read.orc(
    's3://****'
)
print("Second dataframe read with headers set to True")

# df3 = df.join(df2, ['c_0'], "outer")

# df3 = df.join(
#     df2,
#     df["column_test_1"] == df2["column_1"],
#     "outer"
# )

df3 = df.alias('l').join(df2.alias('r'), on='c_0') #.collect()

print("Dataframes have been joined successfully.")
output_file_path = 's3://****'
)

df3.write.orc(
    output_file_path
)
print("Dataframe has been written to csv.")
job.commit()

Ошибка, с которой я сталкиваюсь: 

pyspark.sql.utils.AnalysisException: u'Duplicate column(s): "c_4", "c_38", "c_13", "c_27", "c_50", "c_16", "c_23", "c_24", "c_1", "c_35", "c_30", "c_56", "c_34", "c_7", "c_46", "c_49", "c_57", "c_45", "c_31", "c_53", "c_19", "c_25", "c_10", "c_8", "c_14", "c_42", "c_20", "c_47", "c_36", "c_29", "c_15", "c_43", "c_32", "c_5", "c_37", "c_18", "c_54", "c_3", "__created_at__", "c_51", "c_48", "c_9", "c_21", "c_26", "c_44", "c_55", "c_2", "c_17", "c_40", "c_28", "c_33", "c_41", "c_22", "c_11", "c_12", "c_52", "c_6", "c_39" found, cannot save to file.;'
End of LogType:stdout

Ответы [ 2 ]

0 голосов
/ 11 марта 2019

Я сделал что-то подобное, но в scala , вы можете также конвертировать его в pyspark ...

  • Переименовать имена столбцов в каждом кадре данных 

    dataFrame1.columns.foreach(columnName => {
  dataFrame1 = dataFrame1.select(dataFrame1.columns.head, dataFrame1.columns.tail: _*).withColumnRenamed(columnName, s"left_$columnName")
})

dataFrame1.columns.foreach(columnName => {
  dataFrame2 = dataFrame2.select(dataFrame2.columns.head, dataFrame2.columns.tail: _*).withColumnRenamed(columnName, s"right_$columnName")
})

  • Теперь join, упомянув имена столбцов 

    resultDF = dataframe1.join(dataframe2, dataframe1("left_c_0") === dataframe2("right_c_0"))
0 голосов
/ 11 марта 2019

Здесь нет ярлыка.Pyspark ожидает, что левый и правый фреймы данных будут иметь разные наборы имен полей (за исключением ключа соединения).

Одним из решений будет префикс каждого имени поля с "left_" или "right_" какследует: 

# Obtain columns lists
left_cols = df.columns
right_cols = df2.columns

# Prefix each dataframe's field with "left_" or "right_"
df = df.selectExpr([col + ' as left_' + col for col in left_cols])
df2 = df2.selectExpr([col + ' as right_' + col for col in right_cols])

# Perform join
df3 = df.alias('l').join(df2.alias('r'), on='c_0')
...