pyspark - 如何使用模式匹配从 pyspark 数据框中删除行？

Question

我有一个从 CSV 文件读取的 pyspark 数据帧，该文件具有包含十六进制值的值列。

| date     | part  | feature | value"       |
|----------|-------|---------|--------------|
| 20190503 | par1  | feat2   | 0x0          |
| 20190503 | par1  | feat3   | 0x01         |
| 20190501 | par2  | feat4   | 0x0f32       |
| 20190501 | par5  | feat9   | 0x00         |
| 20190506 | par8  | feat2   | 0x00f45      |
| 20190507 | par1  | feat6   | 0x0e62300000 |
| 20190501 | par11 | feat3   | 0x000000000  |
| 20190501 | par21 | feat5   | 0x03efff     |
| 20190501 | par3  | feat9   | 0x000        |
| 20190501 | par6  | feat5   | 0x000000     |
| 20190506 | par5  | feat8   | 0x034edc45   |
| 20190506 | par8  | feat1   | 0x00000      |
| 20190508 | par3  | feat6   | 0x00000000   |
| 20190503 | par4  | feat3   | 0x0c0deffe21 |
| 20190503 | par6  | feat4   | 0x0000000000 |
| 20190501 | par3  | feat6   | 0x0123fe     |
| 20190501 | par7  | feat4   | 0x00000d0    |

要求是删除包含类似于 0x0、0x00、0x000 等的值的行，这些值在值列中计算为十进制 0(零)。 “0x”之后的 0 数量在整个数据帧中各不相同。通过模式匹配删除是我尝试过的方法，但没有成功。

myFile = sc.textFile("file.txt")
header = myFile.first()

fields = [StructField(field_name, StringType(), True) for field_name in header.split(',')]

myFile_header = myFile.filter(lambda l: "date" in l)
myFile_NoHeader = myFile.subtract(myFile_header)

myFile_df = myFile_NoHeader.map(lambda line: line.split(",")).toDF(schema)

## this is the pattern match I tried 
result = myFile_df.withColumn('Test', regexp_extract(col('value'), '(0x)(0\1*\1*)',2 ))
result.show()

我使用的另一种方法是使用 udf:

def convert_value(x):
    return int(x,16)

在 pyspark 中使用这个 udf 给我

ValueError: invalid literal for int() with base 16: value

Answer 1

我不太明白你的正则表达式，但是当你想匹配所有包含0x0(+任意数量的零)的字符串时，你可以使用^0x0+$。使用正则表达式过滤可以通过 rlike 来实现波浪号否定匹配。

l = [('20190503', 'par1', 'feat2', '0x0'),
('20190503', 'par1', 'feat3', '0x01'),
('20190501', 'par2', 'feat4', '0x0f32'),
('20190501', 'par5', 'feat9', '0x00'),
('20190506', 'par8', 'feat2', '0x00f45'),
('20190507', 'par1', 'feat6', '0x0e62300000'),
('20190501', 'par11', 'feat3', '0x000000000'),
('20190501', 'par21', 'feat5', '0x03efff'),
('20190501', 'par3', 'feat9', '0x000'),
('20190501', 'par6', 'feat5', '0x000000'),
('20190506', 'par5', 'feat8', '0x034edc45'),
('20190506', 'par8', 'feat1', '0x00000'),
('20190508', 'par3', 'feat6', '0x00000000'),
('20190503', 'par4', 'feat3', '0x0c0deffe21'),
('20190503', 'par6', 'feat4', '0x0000000000'),
('20190501', 'par3', 'feat6', '0x0123fe'),
('20190501', 'par7', 'feat4', '0x00000d0')]

columns = ['date', 'part', 'feature', 'value']

df=spark.createDataFrame(l, columns)

expr = "^0x0+$"
df.filter(~ df["value"].rlike(expr)).show()

输出:

+--------+-----+-------+------------+ 
|    date| part|feature|       value| 
+--------+-----+-------+------------+ 
|20190503| par1|  feat3|        0x01| 
|20190501| par2|  feat4|      0x0f32| 
|20190506| par8|  feat2|     0x00f45| 
|20190507| par1|  feat6|0x0e62300000| 
|20190501|par21|  feat5|    0x03efff| 
|20190506| par5|  feat8|  0x034edc45| 
|20190503| par4|  feat3|0x0c0deffe21| 
|20190501| par3|  feat6|    0x0123fe| 
|20190501| par7|  feat4|   0x00000d0| 
+--------+-----+-------+------------+