使用python根据数据文件的数据类型映射表与数据文件？

Posted 2023-04-17

【中文标题】使用python根据数据文件的数据类型映射表与数据文件？

【英文标题】：Mapping table with data files based on data type of the data file using python?

【发布时间】：2018-10-09 06:25:49

【问题描述】：

我有 4 个文本文件，2 个文件包含数据库表的详细信息，其他 2 个包含数据，如下所示。

table1.txt
ename:varchar(10)
eid:smallint(5)
esal:numeric(10,3)

table2.txt
sid:smallint(5)
sname:varchar(10)
sclass:varchar(10)

我的数据文件如下所示，文件名也不是固定的，即它会改变

file1.txt:
aa,1,12222.009
bb,2,12345.012

file2.txt
1,s1,1st_class
2,s2,2nd_class

所以现在我想根据数据的数据类型映射哪个文件匹配将匹配哪个表。我的预期输出应该如下所示。

我的预期输出将在其他日志文件或打印语句中：

table1 matched data file is file2.txt.
table2 matched  data file is file1.txt.

【问题讨论】：

到目前为止你有什么尝试？

我曾尝试通过创建数据框来使用 pyspark，但它没有发生

我实际上.. 我需要两种方式。但python是主要的

如果是 Scala 答案..我可以转换成 python...我正在尝试我知道的两种方式（scala 和 python）

【参考方案1】：

实际上情况并不清楚，因此可能会有所不同。不过我可以给出一些建议。这不是确切的解决方案，但我认为它可以帮助您给出一个想法。

首先我阅读了表格的详细信息；

>>> rdd1 = sc.textFile('/home/ali/table1.txt')
>>> table1 = rdd1.map(lambda x: x.split(':')).map(lambda x: (x[0],x[1])).toDF(['col_name','data_type'])
>>> table1.show()
+--------+-------------+
|col_name|    data_type|
+--------+-------------+
|   ename|  varchar(10)|
|     eid|  smallint(5)|
|    esal|numeric(10,3)|
+--------+-------------+

>>> rdd2 = sc.textFile('/home/ali/table2.txt')
>>> table2 = rdd2.map(lambda x: x.split(':')).map(lambda x: (x[0],x[1])).toDF(['col_name','data_type'])
>>> table2.show()
+--------+-----------+
|col_name|  data_type|
+--------+-----------+
|     sid|smallint(5)|
|   sname|varchar(10)|
|  sclass|varchar(10)|
+--------+-----------+

我读取数据文件，但在此之前您应该定义模式。如果不这样做，所有列的数据类型将默认分配为字符串

>>> from pyspark.sql.types import StructType, StructField, DoubleType, IntegerType, StringType
>>> 
>>> schema1 = StructType([
...     StructField("col1", StringType()),
...     StructField("col2", IntegerType()),
...     StructField("col3", DoubleType())
... ])
>>> 
>>> schema2 = StructType([
...     StructField("col1", IntegerType()),
...     StructField("col2", StringType()),
...     StructField("col3", StringType())
... ])
>>> 
>>> data1 = spark.read.csv('/home/ali/file1.txt', schema=schema1)
>>> data1.show()
+----+----+---------+
|col1|col2|     col3|
+----+----+---------+
|  aa|   1|12222.009|
|  bb|   2|12345.012|
+----+----+---------+

>>> data2 = spark.read.csv('/home/ali/file2.txt', schema=schema2)
>>> data2.show()
+----+----+---------+
|col1|col2|     col3|
+----+----+---------+
|   1|  s1|1st_class|
|   2|  s2|2nd_class|
+----+----+---------+

我定义了一个函数来检查数据类型是否匹配。但是当你定义一个函数时，你应该转换一些数据库数据类型（例如：varchar -> string、numeric -> double ..）我只转换字符串、int 和 double 数据类型。如果您将使用更多数据类型，则应定义所有数据类型

>>> def matchTableData(t,d):
...     matched = []
...     for k1,table in t.items():
...             table_dtypes = []
...             a = True
...             for i in [i.data_type for i in table.select('data_type').collect()]:
...                     if 'char' in i:
...                             table_dtypes.append('string')
...                     elif 'int' in i:
...                             table_dtypes.append('int')
...                     elif 'numeric' in i:
...                             table_dtypes.append('double')
...             for k2,data in d.items():
...                     data_dtypes = [i[1] for i in data.dtypes]
...                     if table_dtypes == data_dtypes:
...                             matched.append([k1,k2])
...     return matched

现在我们准备比较数据类型。我为表和数据创建了两个字典。

>>> tables = 'table1':table1, 'table2':table2
>>> data = 'data1':data1, 'data2':data2
>>> print(matchTableData(tables,data))
[['table1', 'data1'], ['table2', 'data2']]

如您所见，它返回匹配的。正如我之前所说，它可能不是确切的解决方案，但我认为您可以使用其中的一部分

【讨论】：

嗨，感谢您的快速回复，但我无法创建 schema1 硬编码值 schema1 应该动态生成。在上面的答案（schema1，schema2）应该是动态的

您可以为此使用 Pandas。将熊猫导入为 pd data1 = spark.createDataFrame(pd.read_csv('/home/cloudera/ali/file1.txt'))

我可以使用 options(header='false', inferschema='true') 代替 pandas 吗？