Skip to content
Projects
Groups
Snippets
Help
Loading...
Sign in
Toggle navigation
F
ffm-baseline
Project
Project
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
ML
ffm-baseline
Commits
321e3589
Commit
321e3589
authored
6 years ago
by
张彦钊
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
change test file
parent
2b18672a
master
mr/beta/bug22
offic
rtt
updatedb
zhao
zhao22
No related merge requests found
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
14 additions
and
18 deletions
+14
-18
feature_test.py
tensnsorflow/feature_test.py
+14
-18
No files found.
tensnsorflow/feature_test.py
View file @
321e3589
...
...
@@ -214,10 +214,7 @@ def feature_engineer():
app_list_func
(
x
[
11
],
leve2_map
),
app_list_func
(
x
[
12
],
leve2_map
),
[
value_map
[
x
[
0
]],
value_map
[
x
[
13
]],
value_map
[
x
[
14
]],
value_map
[
x
[
15
]],
value_map
[
x
[
16
]],
value_map
[
x
[
17
]],
value_map
[
x
[
18
]],
value_map
[
x
[
19
]],
value_map
[
x
[
20
]],
value_map
[
x
[
21
]],
value_map
[
x
[
22
]],
value_map
[
x
[
23
]],
value_map
[
x
[
24
]],
value_map
[
x
[
25
]],
value_map
[
x
[
26
]]]))
\
.
zipWithIndex
()
.
map
(
lambda
x
:(
x
[
0
][
0
],
x
[
0
][
1
],
x
[
0
][
2
],
x
[
0
][
3
],
x
[
0
][
4
],
x
[
0
][
5
],
x
[
0
][
6
],
x
[
0
][
7
],
x
[
0
][
8
],
x
[
0
][
9
],
x
[
0
][
10
],
x
[
0
][
11
],
x
[
0
][
12
],
x
[
0
][
13
],
x
[
1
]))
value_map
[
x
[
22
]],
value_map
[
x
[
23
]],
value_map
[
x
[
24
]],
value_map
[
x
[
25
]],
value_map
[
x
[
26
]]]))
rdd
.
persist
(
storageLevel
=
StorageLevel
.
MEMORY_ONLY_SER
)
...
...
@@ -225,11 +222,11 @@ def feature_engineer():
train
=
rdd
.
map
(
lambda
x
:
(
x
[
1
],
x
[
2
],
x
[
3
],
x
[
4
],
x
[
5
],
x
[
6
],
x
[
7
],
x
[
8
],
x
[
9
],
x
[
10
],
x
[
11
],
x
[
12
],
x
[
13
]
,
x
[
14
]
))
x
[
10
],
x
[
11
],
x
[
12
],
x
[
13
]))
f
=
time
.
time
()
spark
.
createDataFrame
(
train
)
.
toDF
(
"y"
,
"z"
,
"app_list"
,
"level2_list"
,
"level3_list"
,
"tag1_list"
,
"tag2_list"
,
"tag3_list"
,
"tag4_list"
,
"tag5_list"
,
"tag6_list"
,
"tag7_list"
,
"ids"
,
"number"
)
\
"tag5_list"
,
"tag6_list"
,
"tag7_list"
,
"ids"
)
\
.
repartition
(
1
)
.
write
.
format
(
"tfrecords"
)
.
save
(
path
=
path
+
"tr/"
,
mode
=
"overwrite"
)
h
=
time
.
time
()
print
(
"train tfrecord done"
)
...
...
@@ -242,11 +239,11 @@ def feature_engineer():
test
=
rdd
.
filter
(
lambda
x
:
x
[
0
]
==
validate_date
)
.
map
(
lambda
x
:
(
x
[
1
],
x
[
2
],
x
[
3
],
x
[
4
],
x
[
5
],
x
[
6
],
x
[
7
],
x
[
8
],
x
[
9
],
x
[
10
],
x
[
11
],
x
[
12
],
x
[
13
]
,
x
[
14
]
))
x
[
10
],
x
[
11
],
x
[
12
],
x
[
13
]))
spark
.
createDataFrame
(
test
)
.
toDF
(
"y"
,
"z"
,
"app_list"
,
"level2_list"
,
"level3_list"
,
"tag1_list"
,
"tag2_list"
,
"tag3_list"
,
"tag4_list"
,
"tag5_list"
,
"tag6_list"
,
"tag7_list"
,
"ids"
,
"number"
)
\
"tag5_list"
,
"tag6_list"
,
"tag7_list"
,
"ids"
)
\
.
repartition
(
1
)
.
write
.
format
(
"tfrecords"
)
.
save
(
path
=
path
+
"va/"
,
mode
=
"overwrite"
)
print
(
"va tfrecord done"
)
...
...
@@ -276,7 +273,8 @@ def get_predict(date,value_map,app_list_map,leve2_map,leve3_map):
"left join jerry_test.order_tag ot on e.device_id = ot.device_id "
\
"left join jerry_test.sixin_tag sixin on e.device_id = sixin.device_id "
\
"left join jerry_test.cart_tag cart on e.device_id = cart.device_id "
\
"left join jerry_test.knowledge k on feat.level2 = k.level2_id limit 6000"
"left join jerry_test.knowledge k on feat.level2 = k.level2_id "
\
"where label = 1 limit 60000"
features
=
[
"ucity_id"
,
"ccity_name"
,
"device_type"
,
"manufacturer"
,
"channel"
,
"top"
,
"time"
,
"hospital_id"
,
...
...
@@ -305,21 +303,18 @@ def get_predict(date,value_map,app_list_map,leve2_map,leve3_map):
value_map
.
get
(
x
[
25
],
11
),
value_map
.
get
(
x
[
26
],
12
),
value_map
.
get
(
x
[
27
],
13
),
value_map
.
get
(
x
[
28
],
14
),
value_map
.
get
(
x
[
29
],
15
)
]))
\
.
zipWithIndex
()
.
map
(
lambda
x
:(
x
[
0
][
0
],
x
[
0
][
1
],
x
[
0
][
2
],
x
[
0
][
3
],
x
[
0
][
4
],
x
[
0
][
5
],
x
[
0
][
6
],
x
[
0
][
7
],
x
[
0
][
8
],
x
[
0
][
9
],
x
[
0
][
10
],
x
[
0
][
11
],
x
[
0
][
12
],
x
[
0
][
13
],
x
[
0
][
14
],
x
[
0
][
15
],
x
[
0
][
16
],
x
[
1
]))
]))
rdd
.
persist
(
storageLevel
=
StorageLevel
.
MEMORY_ONLY_SER
)
# native_pre = spark.createDataFrame(rdd.filter(lambda x:x[0] == 0).map(lambda x:(x[3],x[4],x[5])))\
# .toDF("city","uid","cid_id")
# print("native csv")
# native_pre.toPandas().to_csv(local_path
+
"native.csv", header=True)
# native_pre.toPandas().to_csv(local_path
+
"native.csv", header=True)
# spark.createDataFrame(rdd.filter(lambda x: x[0] == 0)
# .map(lambda x: (x[1],x[2],x[6],x[7],x[8],x[9],x[10],x[11],x[12],x[13],x[14],x[15],x[16]
,x[17]
))) \
# .map(lambda x: (x[1],x[2],x[6],x[7],x[8],x[9],x[10],x[11],x[12],x[13],x[14],x[15],x[16]))) \
# .toDF("y","z","app_list", "level2_list", "level3_list","tag1_list", "tag2_list", "tag3_list", "tag4_list",
# "tag5_list", "tag6_list", "tag7_list", "ids"
,"number"
).repartition(1).write.format("tfrecords") \
# "tag5_list", "tag6_list", "tag7_list", "ids").repartition(1).write.format("tfrecords") \
# .save(path=path+"native/", mode="overwrite")
# print("native tfrecord done")
# h = time.time()
...
...
@@ -332,9 +327,9 @@ def get_predict(date,value_map,app_list_map,leve2_map,leve3_map):
spark
.
createDataFrame
(
rdd
.
filter
(
lambda
x
:
x
[
0
]
==
1
)
.
map
(
lambda
x
:
(
x
[
1
],
x
[
2
],
x
[
6
],
x
[
7
],
x
[
8
],
x
[
9
],
x
[
10
],
x
[
11
],
x
[
12
],
x
[
13
],
x
[
14
],
x
[
15
],
x
[
16
]
,
x
[
17
]
)))
\
lambda
x
:
(
x
[
1
],
x
[
2
],
x
[
6
],
x
[
7
],
x
[
8
],
x
[
9
],
x
[
10
],
x
[
11
],
x
[
12
],
x
[
13
],
x
[
14
],
x
[
15
],
x
[
16
])))
\
.
toDF
(
"y"
,
"z"
,
"app_list"
,
"level2_list"
,
"level3_list"
,
"tag1_list"
,
"tag2_list"
,
"tag3_list"
,
"tag4_list"
,
"tag5_list"
,
"tag6_list"
,
"tag7_list"
,
"ids"
,
"number"
)
.
repartition
(
100
)
.
write
.
format
(
"tfrecords"
)
\
"tag5_list"
,
"tag6_list"
,
"tag7_list"
,
"ids"
)
.
repartition
(
100
)
.
write
.
format
(
"tfrecords"
)
\
.
save
(
path
=
path
+
"nearby/"
,
mode
=
"overwrite"
)
print
(
"nearby tfrecord done"
)
...
...
@@ -355,6 +350,7 @@ if __name__ == '__main__':
spark
.
sparkContext
.
setLogLevel
(
"WARN"
)
path
=
"hdfs:///strategy/esmm/"
local_path
=
"/home/gmuser/esmm/"
validate_date
,
value_map
,
app_list_map
,
leve2_map
,
leve3_map
=
feature_engineer
()
get_predict
(
validate_date
,
value_map
,
app_list_map
,
leve2_map
,
leve3_map
)
...
...
This diff is collapsed.
Click to expand it.
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment